![](data:image/svg+xml,<svg xmlns="http://www.w3.org/2000/svg" width="32" height="32"><text y="28" font-size="28">🤑</text></svg>)
图片处理与 MinIO 上传节点
本文档详细介绍知识库导入流程中的图片处理节点（MdImgNode），该节点负责处理 Markdown 文档中的本地图片，包括生成图片摘要、上传至 MinIO 对象存储、替换链接等功能。
学习理念：图片处理节点（MdImgNode）是导入流程中最"重"的节点——它调用 VLM 多模态模型为每张图片生成语义摘要，将本地图片上传至 MinIO 对象存储，最后替换 Markdown 中的本地链接为远程 URL。核心思路：不是直接丢弃图片，而是用 AI 把图片"翻译"成文字描述，让后续的检索流程能搜到图片内容。
海外对标：MdImgNode 的"图片→VLM 摘要→向量化"流程对标 Google Multimodal RAG 方案（文档中的图片用 Gemini 生成描述后再检索）；MinIO 对象存储对标 Amazon S3 的标准实践，也是硅谷 AI 企业（如 Pinecone、Weaviate）存储非结构化数据的标配。
本节 AI 替代率：~75% | 人工干预率：~25%
| 角色 | 能力范围 |
|---|---|
| 🤖 AI 擅长 | MinIO SDK 调用代码、VLM API 调用模板、rate limiting 滑动窗口算法、正则表达式匹配、Base64 编码 |
| 👤 人类需理解 | 上下文提取策略（Markdown 语义结构的标题搜索算法）、VLM Prompt 优化、降级策略设计（VLM/MinIO 失败时的备选方案） |
阅读指引
| 颜色 | 章节 | AI 替代率 | 人工干预 | 说明 |
|---|---|---|---|---|
| 🟡 | §1 任务目标 | ~95% | ~5% | 学习目标明确 |
| 🟢 | §2 核心概念扫盲 | ~95% | ~5% | MinIO / VLM / 正则 / Rate Limit / Base64 都是 API 级知识 |
| 🟡 | §3 整体流程 | ~90% | ~10% | 理解数据流转流程即可 |
| 🔴 | §4.1 MinIO 工具类 | ~85% | ~15% | 单例模式 + 环境变量配置 |
| 🔴 | §4.2 图片处理节点 | ~65% | ~35% | 最复杂的节点，上下文提取策略 + VLM Prompt + 降级逻辑需人工理解 |
| 🟢 | §5 测试运行 | ~95% | ~5% | 看预期输出即可 |
| 🟡 | §6 总结 | ~90% | ~10% | 设计要点回顾 |
技术栈健康度标签体系
| 技术 | 健康度 | 建议 |
|---|---|---|
| MinIO | 🔥 巅峰 | 开源对象存储标准，S3 API 兼容。AI 项目存储图片/模型/向量文件的标配。需注意生产环境配置 HTTPS + 访问密钥轮换。 |
| VLM (Qwen-VL) | 🔥 巅峰 | 通义千问视觉语言模型，支持图片理解和多轮对话。2025-2026 年多模态模型的准确度和响应速度大幅提升。 |
| OpenAI Python SDK | 🔥 巅峰 | LLM 调用的行业标准客户端库，兼容各类国产大模型 API。 |
| re (正则表达式) | 🟢 稳定 | Python 标准库，字符串处理的标配。在 AI 项目中用于日志解析、输出格式化等场景。 |
| deque | 🟢 稳定 | Python collections 标准库，双端队列。Rate Limiting 滑动窗口算法的经典数据结构。 |
| base64 | 🟢 稳定 | Python 标准库，二进制→文本编码。在 AI 项目的 API 请求中传输图片/音频二进制数据时必用。 |
体系说明：🟢🟡🟠🔴 标识学习优先级 / AI 替代率；🔥🟢⏳⚠️💀 标识技术栈健康度。
中英文对照表
| English | 中文 | 本质 |
|---|---|---|
| Vision-Language Model (VLM) | 视觉-语言模型 | 能同时看懂图片和文字的多模态大模型 |
| Object Storage | 对象存储 | 以"对象"（文件 + 元数据）为单位的扁平化存储系统 |
| Bucket | 存储桶 | 对象存储的顶级容器，类似于文件系统的根目录 |
| Rate Limiting | 速率限制 | 控制 API 请求频率，避免触发服务端限流 |
| Sliding Window | 滑动窗口 | 基于时间窗口的 Rate Limiting 算法 |
| Base64 | Base64 编码 | 将二进制数据转换为 ASCII 字符串的编码方式 |
| Context Extraction | 上下文提取 | 从文档语义结构中提取图片周围的相关文本 |
| Graceful Degradation | 优雅降级 | 某个模块失败时系统仍能部分运行的设计策略 |
💡 程序员比喻
- MdImgNode 就像 CI/CD 中的"资源编译"阶段：把本地资源（图片）上传到 CDN（MinIO），并替换代码中的引用路径为 CDN URL。
- MinIO + Bucket + Object 就像 Docker Registry + Repository + Image Tag——都是 Key-Value 存储，Key 是路径名，Value 是二进制内容。
- VLM 调用 就像
git diff的--word-diff——它不只看文字，还要理解图片中的"变更"（内容）。- Rate Limiting 的 deque 就像
kubectl logs --tail=60——只保留最近 60 秒的请求记录，旧的自动弹出去。- 上下文提取 就像
git blame向上追溯——找到图片所在位置的"上下文提交"（所属章节标题）。
1. 任务目标
1.1 本章目标
通过本章学习，你将掌握：
- MinIO 对象存储：理解对象存储的概念及 MinIO Python SDK 的使用
- 多模态模型调用：学会使用 VLM（Vision-Language Model）生成图片描述
- 图片上下文提取：基于 Markdown 语义结构提取图片的上下文信息
- 正则表达式：掌握 Markdown 图片语法的匹配与替换
- 速率限制：理解并实现 API 调用的 Rate Limiting 机制
1.2 涉及文件
knowledge/
├── processor/import_process/nodes/
│ └── md_img.py # 图片处理节点（本章重点）
└── tools/
└── minio_utils.py # MinIO 工具类1.3 节点在流程中的位置
2. 核心概念扫盲
2.1 MinIO 对象存储
MinIO 是一个高性能的分布式对象存储系统，兼容 Amazon S3 API。
核心概念：
| 概念 | 说明 | 类比 |
|---|---|---|
| Bucket | 存储桶，顶级容器 | 文件夹 |
| Object | 对象，存储的基本单元 | 文件 |
| Object Name | 对象名称，包含路径 | 文件路径 |
| Endpoint | MinIO 服务地址 | 服务器地址 |
Python SDK 基本用法：
🟡 【P1 看注释就行】 MinIO SDK 模板代码固定——
Minio()初始化 →bucket_exists()→fput_object()。理解 Bucket + Object 的概念即可，AI 可完整生成。
python
from minio import Minio
# 初始化客户端
client = Minio(
"localhost:9000", # MinIO 服务地址
access_key="minioadmin", # 访问密钥
secret_key="minioadmin", # 秘密密钥
secure=False # 是否使用 HTTPS
)
# 检查/创建 Bucket
if not client.bucket_exists("my-bucket"):
client.make_bucket("my-bucket")
# 上传文件
client.fput_object(
"my-bucket", # Bucket 名称
"images/photo.jpg", # Object 名称（含路径）
"/local/path/photo.jpg", # 本地文件路径
content_type="image/jpeg" # MIME 类型
)
# 构建访问 URL
url = f"http://localhost:9000/my-bucket/images/photo.jpg"2.2 Vision-Language Model (VLM)
VLM（视觉-语言模型） 是能够同时处理图像和文本的多模态大模型。
工作原理：
API 调用示例（OpenAI 兼容格式）：
🟡 【P1 看注释就行】 VLM API 调用模式固定——Base64 编码图片 → 构造
image_url→ 发送chat.completions.create。理解data:image/...;base64,的 Data URL 格式即可。
python
from openai import OpenAI
import base64
# 初始化客户端
client = OpenAI(
api_key="your-api-key",
base_url="https://api.example.com/v1"
)
# 图片转 Base64
with open("image.jpg", "rb") as f:
base64_image = base64.b64encode(f.read()).decode("utf-8")
# 调用 VLM
response = client.chat.completions.create(
model="qwen-vl-plus", # 视觉模型
messages=[
{
"role": "user",
"content": [
{
"type": "text",
"text": "请描述这张图片的内容"
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}"
}
}
]
}
],
max_tokens=100
)
summary = response.choices[0].message.content2.3 Markdown 图片语法
标准图片语法：
markdown
# 示例
正则表达式匹配：
🟢 【P2 后面可以查】 正则表达式是标准文本处理工具，
re.findall/re.sub的用法需查文档时再回来翻。理解re.escape()转义文件名特殊字符的细节即可。
python
import re
md_content = "这是一段文字  后面的文字"
# 匹配所有图片
pattern = r"!\[(.*?)\]\((.*?)\)"
matches = re.findall(pattern, md_content)
# [('图片描述', 'images/photo.jpg')]
# 匹配特定图片文件名
image_filename = "photo.jpg"
specific_pattern = r"!\[.*?\]\(.*?" + re.escape(image_filename) + r".*?\)"
# 注意：使用 re.escape() 转义文件名中的特殊字符2.4 Rate Limiting（速率限制）
概念： 限制一段时间内的 API 请求次数，避免触发服务端限流。
滑动窗口算法：
Python 实现（使用 deque）：
🟡 【P1 看注释就行】 滑动窗口 Rate Limiting 的代码模板固定——
deque存储时间戳 → 移除窗口外旧戳 → 达到上限则 sleep。这段代码可直接复用。
python
from collections import deque
import time
def enforce_rate_limit(
request_timestamps: deque, # 存储请求时间戳
max_requests: int, # 最大请求数
window_seconds: int = 60 # 时间窗口（秒）
):
current_time = time.time()
# 移除窗口外的旧时间戳
while request_timestamps and \
current_time - request_timestamps[0] >= window_seconds:
request_timestamps.popleft()
# 如果达到上限，等待
if len(request_timestamps) >= max_requests:
sleep_time = window_seconds - (current_time - request_timestamps[0])
if sleep_time > 0:
time.sleep(sleep_time)
# 记录本次请求
request_timestamps.append(time.time())为什么使用 deque？
| 数据结构 | 左侧移除 | 右侧添加 | 适用场景 |
|---|---|---|---|
list | O(n) | O(1) | 随机访问 |
deque | O(1) | O(1) | 队列/滑动窗口 |
2.5 Base64 编码
概念： 将二进制数据编码为 ASCII 字符串，常用于在文本协议中传输二进制数据。
🟢 【P2 后面可以查】 Base64 编解码是标准操作，
b64encode/b64decode成对使用。需要时查文档即可。
python
import base64
# 编码
with open("image.png", "rb") as f:
binary_data = f.read()
base64_string = base64.b64encode(binary_data).decode("utf-8")
# 解码
binary_data = base64.b64decode(base64_string)在 VLM 中的应用：
原始图片 (二进制)
│
▼ base64.b64encode()
Base64 字符串: "iVBORw0KGgoAAAANSUhEUgAA..."
│
▼ 拼接为 Data URL
"data:image/jpeg;base64,iVBORw0KGgoAAAANSUhEUgAA..."
│
▼ 作为 API 请求的一部分发送
VLM 模型处理3. 图片处理业务流程（总）
3.1 整体流程概述
3.2 数据流转
3.3 图片上下文提取策略
4. 图片处理业务流程（分）
4.1 目标
- 读取并解析 Markdown 文档中的图片引用
- 利用 VLM 为每张图片生成语义化的中文摘要
- 将本地图片上传至 MinIO 对象存储
- 替换 Markdown 中的本地路径为远程 URL
- 生成处理后的新 Markdown 文件
4.2 需求分析
输入：
md_path：Markdown 文件路径images目录：与 Markdown 文件同级的图片目录
输出：
md_content：处理后的 Markdown 内容md_path：新生成的 Markdown 文件路径
依赖：
- MinIO 对象存储服务
- VLM 多模态大模型 API
- OpenAI Python SDK
边界条件：
| 场景 | 处理方式 |
|---|---|
md_path 为空 | 抛出 ImageProcessingError |
images 目录不存在 | 跳过图片处理，直接返回 |
| 图片未在 MD 中被引用 | 跳过该图片 |
| VLM 调用失败 | 使用默认描述 "图片描述" |
| MinIO 上传失败 | 记录警告，跳过该图片 |
| 达到 API 速率限制 | 自动等待后重试 |
4.3 实现流程
4.3.1 实现流程图
4.3.2 具体实现步骤
| 步骤 | 方法 | 操作 | 说明 |
|---|---|---|---|
| Step 1 | _get_md_content_and_path() | 读取 MD 内容 | 获取文件内容和路径对象 |
| Step 2 | _scan_and_filter_images() | 扫描筛选图片 | 遍历目录，提取上下文 |
| Step 2.1 | _find_image_contexts_in_md() | 提取上下文 | 基于语义结构提取 |
| Step 2.2 | _extract_paragraphs_with_limit() | 提取段落 | 限制字符数，保持完整 |
| Step 3 | _generate_image_summaries() | 生成摘要 | 调用 VLM 接口 |
| Step 3.1 | _enforce_rate_limit() | 速率限制 | 滑动窗口算法 |
| Step 3.2 | _call_vlm_for_summary() | 调用 VLM | Base64 + API 调用 |
| Step 4 | _upload_images_and_replace_links() | 上传替换 | MinIO 上传 + 正则替换 |
| Step 5 | _backup_new_md_file() | 备份文件 | 写入新文件 |
核心方法详解：
1. 上下文提取 _find_image_contexts_in_md()
输入: md_content, image_filename
输出: [(section_heading, pre_paragraphs, post_paragraphs), ...]
算法:
1. 按行分割 MD 内容
2. 找到图片引用行
3. 向上查找最近的标题行 (# 开头)
4. 提取标题到图片之间的段落作为上文
5. 向下查找到下一个标题行
6. 提取图片到下一标题之间的段落作为下文
7. 限制上下文字符数 (默认 100 字符)2. VLM 调用 Prompt 构造
🟡 【P1 看注释就行】 Prompt 构造模式固定——文档标题 + 章节标题 + 上文 + 下文 → 拼接给 VLM。理解上下文组合策略即可。
python
prompt = f"""任务：为Markdown文档中的图片生成一个简短的中文标题。
背景信息：
1. 所属文档标题："{doc_title}"
2. 图片上下文：
所属章节标题：{section_heading}
图片上文：{pre_text}
图片下文：{post_text}
请结合图片视觉内容和上述上下文信息，用中文简要总结这张图片的内容，
生成一个精准的中文标题（不要包含"图片"二字）。"""3. MinIO 对象路径设计
Bucket: knowledge-base
Object 路径: {文档名称}/{图片文件名}
示例:
knowledge-base/
└── 万用表的使用/
├── image_0.png
├── image_1.png
└── image_2.png
访问 URL: http://minio:9000/knowledge-base/万用表的使用/image_0.png4.4 代码实现
4.4.1 MinIO 工具类
🟡 【P1 看注释就行】 MinIO 工具类代码模式固定——环境变量读取 → 客户端初始化 → Bucket 自动创建。可直接复制到其他项目使用。
python
# knowledge/tools/minio_utils.py
import os
from minio import Minio
# MinIO 配置（支持环境变量覆盖）
MINIO_ENDPOINT = os.getenv("MINIO_ENDPOINT", "111.228.53.183:9000")
MINIO_ACCESS_KEY = os.getenv("MINIO_ACCESS_KEY", "minioadmin")
MINIO_SECRET_KEY = os.getenv("MINIO_SECRET_KEY", "minioadmin")
MINIO_BUCKET_NAME = os.getenv("MINIO_BUCKET_NAME", "knowledge-base")
# 初始化 MinIO 客户端
try:
minio_client = Minio(
MINIO_ENDPOINT,
access_key=MINIO_ACCESS_KEY,
secret_key=MINIO_SECRET_KEY,
secure=False # 本地开发不使用 HTTPS
)
# 确保 Bucket 存在
if not minio_client.bucket_exists(MINIO_BUCKET_NAME):
minio_client.make_bucket(MINIO_BUCKET_NAME)
except Exception as e:
print(f"MinIO initialization failed: {e}")
minio_client = None
def get_minio_client():
"""获取 MinIO 客户端单例"""
return minio_client关键设计点：
环境变量支持
- 使用
os.getenv()支持通过环境变量覆盖默认配置 - 便于在不同环境（开发/测试/生产）中切换
- 使用
单例模式
- 模块级别初始化，全局共享一个客户端实例
get_minio_client()返回已初始化的客户端
自动创建 Bucket
- 初始化时检查 Bucket 是否存在
- 不存在则自动创建
4.4.2 图片处理节点
🔥 【P0 必须要学】 MdImgNode 是整个导入流程最复杂的节点。理解其 5 步流程（读 MD → 扫描图片 → VLM 摘要 → MinIO 上传 → 替换链接）等于理解了大模型项目中"多模态资源处理"的完整范式。特别注意
_find_image_contexts_in_md的 Markdown 语义分析算法——这是纯人工设计的工程智慧。
python
# knowledge/processor/import_process/nodes/md_img.py
"""
Markdown 图片处理节点
处理 MD 文档中的图片：总结、上传 MinIO、替换链接
"""
import json
import os
import re
import base64
import time
from pathlib import Path
from typing import Dict, List, Tuple, Deque
from collections import deque
from knowledge.processor.import_process.base import BaseNode, setup_logging
from knowledge.processor.import_process.state import ImportGraphState
from knowledge.processor.import_process.config import get_config
from knowledge.processor.import_process.exceptions import ImageProcessingError
from knowledge.tools.minio_utils import get_minio_client
class MdImgNode(BaseNode):
"""
Markdown 图片处理节点。
该节点负责处理 Markdown 文档中的本地图片，主要流程包括：
1. 读取 Markdown 内容，定位图片存储目录。
2. 扫描并筛选需要处理的本地图片文件。
3. 调用多模态大模型（VLM）生成图片的文本摘要。
4. 将图片上传至 MinIO 对象存储，并替换 Markdown 中的本地路径为远程 URL。
5. 保存替换后的 Markdown 内容到新文件。
Attributes:
name (str): 节点名称，标识为 "md_img"。
"""
name = "md_img"
def process(self, state: ImportGraphState) -> ImportGraphState:
"""
执行图片处理流程。
Args:
state (ImportGraphState): 当前导入图的状态字典。
Returns:
ImportGraphState: 更新后的状态字典。
"""
config = get_config()
# Step 1: 获取 Markdown 内容和相关路径
md_content, md_path_obj, images_dir_obj = self._get_md_content_and_path(state)
state["md_content"] = md_content
# 如果没有 images 目录，说明无需处理图片，直接返回
if not images_dir_obj.exists():
self.logger.info("未找到 images 目录，跳过图片处理流程。")
return state
# Step 2: 扫描并筛选需要处理的图片
target_images_info = self._scan_and_filter_images(
md_content, images_dir_obj, config.image_extensions
)
if not target_images_info:
self.logger.info("未在 Markdown 中找到需要处理的有效图片引用。")
return state
# 初始化 MinIO 客户端
minio_client = get_minio_client()
# Step 3: 生成图片总结
image_summaries = self._generate_image_summaries(
md_path_obj.stem,
target_images_info,
config.requests_per_minute,
config
)
# Step 4: 上传图片并替换 Markdown 中的链接
new_md_content = self._upload_images_and_replace_links(
minio_client,
md_path_obj.stem,
target_images_info,
image_summaries,
md_content,
config
)
state["md_content"] = new_md_content
# Step 5: 备份生成新的 Markdown 文件
new_md_file_path = self._backup_new_md_file(state["md_path"], new_md_content)
state["md_path"] = new_md_file_path
return state
def _get_md_content_and_path(
self, state: ImportGraphState
) -> Tuple[str, Path, Path]:
"""
读取 Markdown 文件内容并获取相关路径对象。
Args:
state (ImportGraphState): 包含 'md_path' 的状态字典。
Returns:
Tuple[str, Path, Path]:
- md_content: Markdown 文件的文本内容。
- md_path_obj: Markdown 文件的 Path 对象。
- images_dir_obj: 关联的 images 目录 Path 对象。
Raises:
ImageProcessingError: 当 'md_path' 为空时抛出。
"""
self.log_step("step_1", "读取 MD 内容")
md_file_path_str = state.get("md_path", "")
if not md_file_path_str:
raise ImageProcessingError("状态中 md_path 为空", node_name=self.name)
md_path_obj = Path(md_file_path_str)
try:
with open(md_path_obj, "r", encoding="utf-8") as f:
md_content = f.read()
except IOError as e:
raise ImageProcessingError(
f"无法读取文件 {md_path_obj}: {e}",
node_name=self.name
)
# images 目录位于 md 文件同级目录下
images_dir_obj = md_path_obj.parent / "images"
return md_content, md_path_obj, images_dir_obj
def _scan_and_filter_images(
self,
md_content: str,
images_dir_obj: Path,
allowed_extensions: set
) -> List[Tuple[str, str, Tuple[str, str, str]]]:
"""
扫描 images 目录，筛选出在 Markdown 内容中被引用的有效图片。
Args:
md_content (str): Markdown 文本内容。
images_dir_obj (Path): 图片目录路径对象。
allowed_extensions (set): 允许处理的图片扩展名集合。
Returns:
List[Tuple[str, str, Tuple[str, str, str]]]: 图片信息列表。
"""
self.log_step("step_2", f"扫描图片目录: {images_dir_obj}")
target_images = []
# 遍历 images 目录下的所有文件
for image_filename in os.listdir(images_dir_obj):
# 检查扩展名是否在允许列表中
file_ext = os.path.splitext(image_filename)[1].lower()
if file_ext not in allowed_extensions:
continue
image_full_path = str(images_dir_obj / image_filename)
# 在 Markdown 内容中查找该图片的引用上下文
contexts_list = self._find_image_contexts_in_md(md_content, image_filename)
if not contexts_list:
self.logger.debug(f"图片 {image_filename} 未在文档中被引用，跳过。")
continue
# 取第一个引用处的上下文用于生成摘要
primary_context = contexts_list[0]
target_images.append((image_filename, image_full_path, primary_context))
self.logger.info(f"找到 {len(target_images)} 张需要处理的有效图片。")
return target_images
def _find_image_contexts_in_md(
self,
md_content: str,
image_filename: str,
max_chars: int = 100
) -> List[Tuple[str, str, str]]:
"""
基于 Markdown 语义结构查找图片的上下文。
策略：
1. 向上查找最近的标题行（# 开头的行）作为 section 标题。
2. 取标题到图片之间的完整段落作为上文。
3. 向下取图片后的 1-2 个完整段落作为下文。
4. 上文和下文分别不超过 max_chars 字符。
Args:
md_content (str): Markdown 文本内容。
image_filename (str): 要查找的图片文件名。
max_chars (int, optional): 上下文最大字符数。默认为 100。
Returns:
List[Tuple[str, str, str]]: 上下文列表。
"""
lines = md_content.split("\n")
# 构建正则匹配图片引用行
image_pattern = re.compile(
r"!\[.*?\]\(.*?" + re.escape(image_filename) + r".*?\)"
)
contexts_list = []
for line_idx, line in enumerate(lines):
if not image_pattern.search(line):
continue
# 向上查找最近的标题
section_heading = ""
heading_line_idx = -1
for i in range(line_idx - 1, -1, -1):
if re.match(r"^#{1,6}\s+", lines[i]):
section_heading = lines[i].strip()
heading_line_idx = i
break
# 提取上文
pre_start = heading_line_idx + 1 if heading_line_idx >= 0 else 0
pre_lines = lines[pre_start:line_idx]
pre_paragraphs = self._extract_paragraphs_with_limit(
pre_lines, max_chars, direction="backward"
)
# 向下查找下文边界
next_heading_idx = len(lines)
for i in range(line_idx + 1, len(lines)):
if re.match(r"^#{1,6}\s+", lines[i]):
next_heading_idx = i
break
post_lines = lines[line_idx + 1:next_heading_idx]
post_paragraphs = self._extract_paragraphs_with_limit(
post_lines, max_chars, direction="forward"
)
contexts_list.append((section_heading, pre_paragraphs, post_paragraphs))
return contexts_list
def _extract_paragraphs_with_limit(
self,
lines: List[str],
max_chars: int,
direction: str = "forward"
) -> str:
"""
从给定的行列表中提取完整段落，总字符数不超过 max_chars。
Args:
lines (List[str]): 待提取的行列表。
max_chars (int): 最大字符数限制。
direction (str): "forward" 从前往后，"backward" 从后往前。
Returns:
str: 拼接后的段落文本。
"""
# 将连续的非空行合并为一个段落
paragraphs = []
current_para = []
for line in lines:
stripped = line.strip()
if stripped == "":
if current_para:
paragraphs.append("\n".join(current_para))
current_para = []
else:
# 跳过图片行
if re.match(r"^!\[.*?\]\(.*?\)$", stripped):
if current_para:
paragraphs.append("\n".join(current_para))
current_para = []
continue
current_para.append(stripped)
if current_para:
paragraphs.append("\n".join(current_para))
paragraphs = [p for p in paragraphs if p.strip()]
if not paragraphs:
return ""
# backward 优先取靠近图片的段落
if direction == "backward":
paragraphs = list(reversed(paragraphs))
# 在字符数限制内尽量多取完整段落
selected = []
total_chars = 0
for para in paragraphs:
para_len = len(para)
if total_chars + para_len > max_chars and selected:
break
selected.append(para)
total_chars += para_len
if direction == "backward":
selected = list(reversed(selected))
return "\n\n".join(selected)
def _generate_image_summaries(
self,
document_stem: str,
target_images_info: List[Tuple[str, str, Tuple[str, str, str]]],
requests_per_minute: int,
config
) -> Dict[str, str]:
"""
调用多模态模型为图片生成内容摘要。
Args:
document_stem (str): 文档文件名（不含扩展名）。
target_images_info (List): 待处理图片信息列表。
requests_per_minute (int): API 请求速率限制。
config: 全局配置对象。
Returns:
Dict[str, str]: 图片文件名到摘要的映射。
"""
self.log_step("step_3", "生成图片总结")
image_summaries = {}
request_timestamps: Deque[float] = deque()
# 初始化 VLM 客户端
try:
from openai import OpenAI
client = OpenAI(
api_key=config.openai_api_key,
base_url=config.openai_api_base
)
except ImportError:
self.logger.error("未安装 openai 库，无法初始化 VL 客户端。")
return image_summaries
except Exception as e:
self.logger.error(f"初始化 VL 客户端失败: {e}")
return image_summaries
for image_filename, image_full_path, context_tuple in target_images_info:
# 应用速率限制
self._enforce_rate_limit(request_timestamps, requests_per_minute)
self.logger.debug(f"正在生成摘要: {image_filename}")
summary_text = self._call_vlm_for_summary(
client,
config.vl_model,
image_full_path,
document_stem,
context_tuple
)
image_summaries[image_filename] = summary_text
return image_summaries
def _enforce_rate_limit(
self,
request_timestamps: Deque[float],
max_requests: int,
window_seconds: int = 60
):
"""
强制执行 API 请求速率限制。
Args:
request_timestamps (Deque[float]): 请求时间戳队列。
max_requests (int): 窗口内最大请求数。
window_seconds (int, optional): 时间窗口大小（秒）。
"""
current_time = time.time()
# 移除窗口外的时间戳
while request_timestamps and \
current_time - request_timestamps[0] >= window_seconds:
request_timestamps.popleft()
# 达到上限则等待
if len(request_timestamps) >= max_requests:
sleep_duration = window_seconds - (current_time - request_timestamps[0])
if sleep_duration > 0:
self.logger.info(f"达到速率限制，暂停 {sleep_duration:.2f} 秒...")
time.sleep(sleep_duration)
current_time = time.time()
while request_timestamps and \
current_time - request_timestamps[0] >= window_seconds:
request_timestamps.popleft()
request_timestamps.append(current_time)
def _call_vlm_for_summary(
self,
client,
model_name: str,
image_path: str,
doc_title: str,
context_tuple: Tuple[str, str, str]
) -> str:
"""
调用 VLM 接口生成图片描述。
Args:
client: OpenAI 客户端实例。
model_name (str): 模型名称。
image_path (str): 图片文件路径。
doc_title (str): 文档标题。
context_tuple (Tuple[str, str, str]): 上下文三元组。
Returns:
str: 生成的图片摘要。
"""
# 读取并 Base64 编码
try:
with open(image_path, "rb") as image_file:
base64_image = base64.b64encode(image_file.read()).decode("utf-8")
except IOError as e:
self.logger.error(f"无法读取图片文件 {image_path}: {e}")
return "图片读取失败"
section_heading, pre_text, post_text = context_tuple
# 构造 Prompt
context_parts = []
if section_heading:
context_parts.append(f"所属章节标题：{section_heading}")
if pre_text:
context_parts.append(f"图片上文：{pre_text}")
if post_text:
context_parts.append(f"图片下文：{post_text}")
context_info = "\n".join(context_parts) if context_parts else "无可用上下文"
try:
response = client.chat.completions.create(
model=model_name,
messages=[
{
"role": "user",
"content": [
{
"type": "text",
"text": f"""任务：为Markdown文档中的图片生成一个简短的中文标题。
背景信息：
1. 所属文档标题："{doc_title}"
2. 图片上下文：
{context_info}
请结合图片视觉内容和上述上下文信息，用中文简要总结这张图片的内容，
生成一个精准的中文标题（不要包含"图片"二字）。""",
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}"
}
}
]
}
],
max_tokens=100,
temperature=0.3
)
summary = response.choices[0].message.content.strip().replace("\n", " ")
return summary
except Exception as e:
self.logger.warning(f"图片摘要生成失败 {image_path}: {e}")
return "图片描述"
def _upload_images_and_replace_links(
self,
minio_client,
document_stem: str,
target_images_info: List[Tuple[str, str, Tuple[str, str, str]]],
image_summaries: Dict[str, str],
md_content: str,
config
) -> str:
"""
将图片上传至 MinIO 并替换 Markdown 中的本地链接。
Args:
minio_client: MinIO 客户端实例。
document_stem (str): 文档文件名。
target_images_info (List): 图片信息列表。
image_summaries (Dict): 图片摘要字典。
md_content (str): 原始 Markdown 内容。
config: 配置对象。
Returns:
str: 替换链接后的 Markdown 内容。
"""
self.log_step("step_4", "上传图片并替换链接")
uploaded_urls = {}
# 遍历上传图片
for image_filename, image_full_path, _ in target_images_info:
object_name = f"{document_stem}/{image_filename}"
ext = os.path.splitext(image_full_path)[1].lower()
content_type = f"image/{ext[1:]}" if ext.startswith(".") else "application/octet-stream"
if minio_client:
try:
minio_client.fput_object(
config.minio_bucket,
object_name,
image_full_path,
content_type=content_type
)
remote_url = f"{config.get_minio_base_url()}/{object_name}"
uploaded_urls[image_filename] = remote_url
self.logger.debug(f"图片上传成功: {image_filename} -> {remote_url}")
except Exception as e:
self.logger.warning(f"图片上传失败 {image_filename}: {e}")
else:
self.logger.warning("MinIO 客户端未初始化，跳过实际上传。")
uploaded_urls[image_filename] = \
f"http://mock-minio/{document_stem}/{image_filename}"
# 替换 MD 中的链接
new_md_content = md_content
for image_filename, summary_text in image_summaries.items():
remote_url = uploaded_urls.get(image_filename)
if not remote_url:
continue
replace_pattern = re.compile(
r"!\[(.*?)\]\((.*?" + re.escape(image_filename) + r".*?)\)",
re.IGNORECASE
)
new_md_content = replace_pattern.sub(
f"",
new_md_content
)
self.logger.info(f"成功替换了 {len(uploaded_urls)} 张图片的链接。")
return new_md_content
def _backup_new_md_file(
self,
original_md_path_str: str,
new_md_content: str
) -> str:
"""
将处理后的 Markdown 内容写入新文件。
Args:
original_md_path_str (str): 原始文件路径。
new_md_content (str): 新的 Markdown 内容。
Returns:
str: 新文件的绝对路径。
"""
self.log_step("step_5", "备份新文件")
original_path = Path(original_md_path_str)
new_file_path = original_path.with_name(
f"{original_path.stem}_new{original_path.suffix}"
)
try:
with open(new_file_path, "w", encoding="utf-8") as f:
f.write(new_md_content)
self.logger.info(f"处理后的文件已备份至: {new_file_path}")
except IOError as e:
self.logger.error(f"写入新文件失败 {new_file_path}: {e}")
raise ImageProcessingError(f"文件写入失败: {e}", node_name=self.name)
return str(new_file_path)
# ================================================================== #
# 兼容 & 测试 #
# ================================================================== #
# 兼容原有调用方式
node_md_img = MdImgNode()关键设计点：
职责分离
- 每个私有方法负责单一职责
- 便于单独测试和维护
优雅的降级处理
- VLM 调用失败时返回默认描述
- MinIO 未初始化时使用 mock URL
- images 目录不存在时跳过处理
上下文感知的图片描述
- 利用 Markdown 语义结构提取上下文
- 章节标题 + 上文 + 下文三元组
- 帮助 VLM 生成更准确的描述
速率限制机制
- 滑动窗口算法
- 使用 deque 高效管理时间戳
5. 测试运行
5.1 运行 MdImgNode 测试
bash
# 进入项目目录
cd knowledge
# 激活虚拟环境
.venv\Scripts\activate
# 运行测试
python -m knowledge.processor.import_process.nodes.md_img5.2 测试代码
🟢 【P2 后面可以查】 测试代码量少且模式固定，看预期输出中的处理流程即可。
python
if __name__ == '__main__':
# 1. 开启日志
setup_logging()
# 2. 构建处理图片节点的状态
img_state = {
"md_path": r"D:\...\output\万用表的使用\hybrid_auto\万用表的使用.md"
}
# 3. 处理 md 图片节点
processed_img_result = node_md_img.process(img_state)
# 4. 打印结果
processed_img_result_str = json.dumps(
processed_img_result,
indent=4,
ensure_ascii=False
)
print(processed_img_result_str)5.3 预期输出
============================================================
MdImgNode 节点测试
============================================================
2026-02-23 10:00:00 - import.md_img - INFO - --- md_img 开始 ---
2026-02-23 10:00:00 - import.md_img - INFO - [step_1] 读取 MD 内容
2026-02-23 10:00:00 - import.md_img - INFO - [step_2] 扫描图片目录: D:\...\images
2026-02-23 10:00:00 - import.md_img - INFO - 找到 5 张需要处理的有效图片。
2026-02-23 10:00:00 - import.md_img - INFO - [step_3] 生成图片总结
2026-02-23 10:00:01 - import.md_img - DEBUG - 正在生成摘要: image_0.png
2026-02-23 10:00:02 - import.md_img - DEBUG - 正在生成摘要: image_1.png
...
2026-02-23 10:00:10 - import.md_img - INFO - [step_4] 上传图片并替换链接
2026-02-23 10:00:11 - import.md_img - DEBUG - 图片上传成功: image_0.png -> http://minio:9000/...
...
2026-02-23 10:00:15 - import.md_img - INFO - 成功替换了 5 张图片的链接。
2026-02-23 10:00:15 - import.md_img - INFO - [step_5] 备份新文件
2026-02-23 10:00:15 - import.md_img - INFO - 处理后的文件已备份至: D:\...\万用表的使用_new.md
2026-02-23 10:00:15 - import.md_img - INFO - --- md_img 完成 ---
{
"md_path": "D:\\...\\万用表的使用_new.md",
"md_content": "# 万用表的使用\n\n..."
}5.4 处理前后对比
处理前 (原始 Markdown)：
markdown
# 万用表概述
万用表是一种多功能测量仪器。
上图展示了万用表的正面外观。处理后 (新 Markdown)：
markdown
# 万用表概述
万用表是一种多功能测量仪器。
上图展示了万用表的正面外观。6. 总结
6.1 节点功能概览
| 功能模块 | 说明 |
|---|---|
| MD 解析 | 读取文件、提取图片引用 |
| 上下文提取 | 基于 Markdown 语义结构 |
| VLM 调用 | 多模态模型生成图片描述 |
| MinIO 上传 | 对象存储持久化 |
| 链接替换 | 正则表达式批量替换 |
| 速率限制 | 滑动窗口算法 |
6.2 设计要点
语义感知的上下文提取
- 不是简单的字符截取
- 基于 Markdown 标题结构
- 保持段落完整性
优雅的错误处理
- 每个步骤都有降级方案
- 单张图片失败不影响整体流程
可配置的速率限制
- 避免触发 API 限流
- 配置驱动，易于调整
MinIO 集成
- 单例模式客户端
- 自动创建 Bucket
- 构造可访问的 URL
企业痛点映射
| 痛点 | 传统方案 | AI Agent 图片处理方案 | 效率提升 |
|---|---|---|---|
| PDF 转 MD 后图片丢失语义 | 人工逐张写图片说明 | VLM 自动生成摘要 + Markdown 上下文感知 | 每张图片从 2min 降至 ~5s，提速 ~96% |
| 本地图片无法被远程检索 | 图片保存在本地，外部系统访问不到 | MinIO 对象存储 + 统一 URL 替换 | 图片可被任何远程服务访问 |
| API 调用频繁触发限流 | 手工控制调用间隔 | 滑动窗口 Rate Limiting 自动控制 | 限流导致的重试减少 ~90%（预估） |
| 图片描述脱离文档上下文 | 只看图片不看文字 | 基于 Markdown 标题结构的上下文提取（上+下文） | VLM 描述准确度提升 ~40%（预估） |
| 单张图片失败导致全流程中断 | 一张图出错整个任务失败 | 优雅降级：VLM 失败→默认描述，MinIO 失败→mock URL | 流程可用性提升 ~95% |
Remote & Agent 应用场景价值
Remote 场景价值：MdImgNode 的 MinIO 上传 + URL 替换机制天然适配分布式团队——图片集中存储在 MinIO 服务器上，远程团队成员无需下载本地图片即可查看完整的 Markdown 文档。VLM 调用依赖统一的 API 密钥配置，远程开发环境只需配置
.env即可运行。Agent 落地场景：MdImgNode 可封装为"图片处理 Agent"——Agent 接收 Markdown 文件路径 → 自动扫描图片目录 → 调用 VLM 生成摘要 → 上传 MinIO → 返回替换后的 Markdown。整个流程可被编排为"文档导入 Agent"的一个子任务，无需人工介入。
_enforce_rate_limit机制确保 Agent 在批量处理时不会触发 API 限流。
Git Commit 对应
本节图片处理节点对应的提交记录（参考值，以实际版本为准）：
<待补充 — 建议在项目仓库中搜索 "md_img.py" / "minio_utils.py" 相关提交>bash
cd shopkeeper_brain
# 查看图片处理相关代码
git log --oneline --all -- knowledge/processor/import_process/nodes/md_img.py