实现图片处理节点 node_md_img，用视觉模型识别图片内容、上传 MinIO 替换为网络地址，并完成对象存储与 LLM 的 infra 封装。

本篇要点

  • 梳理当前图的节点状态，明确图片处理要解决的问题
  • 实现 node_md_img：scan_images 取图片上下文 → 视觉模型 summarize_images → 上传 MinIO 并替换 md 内容
  • 掌握 MinIO 的 docker 部署、可视化操作、桶权限 policy 与 Python 增删查
  • 用 MinioGateway 与 LLMProvider 把对象存储和模型能力封装进 infra
  • 回顾正则：re.escape、search/match/findall/finditer、贪婪与非贪婪

内容回顾

1、上一天主题（Day02）

  • 定义 ImportGraphState(TypedDict)，共 12 个属性，配套 create_default_state / get_default_state
  • 节点与业务分层：@node_log 装饰器 + add_running_task / add_done_task
  • 主图构建：builder → add_node → set_entry_point → 条件边 → compile
  • 任务分发节点 node_entry，PDF 解析节点 node_pdf_to_md（MinerU 上传/轮询/下载解压重命名）

2、基础工具小结

  • Path 属性：.name、.stem、.suffix、.parent、.parents
  • Path 方法：.is_dir、.exists、.mk_dir、.read_text、.read_bytes、write_text、write_bytes、.rename、.with_name、.rglob、.glob
  • shutil 方法：rmtree(文件夹)、unpack_archive(解压的文件, 目标文件夹)、make_archive("压缩文件名","压缩格式",压缩文件或者文件夹地址)
  • requests：请求用 requests.get/post/put/delete(url, headers, params, json, data, timeout)；响应用 .status_code、.json()、.text、.content，以及 requests.Session()
  • MinerU 的使用：申请上传地址和 batch_id（POST）→ 向指定的预签名地址请求 → 轮询获取响应结果

今天内容

1、当前节点状态

项目主图目前已串起两个节点，节点状态如下：

1
2
3
4
5
6
                  ------------------->
node_entry                     ----> node_md_img
               ---> node_pdf_to_md

node_pdf_to_md : pdf -> md -> 本地存储 -> state -> md_path
node_md_img    : md_path -> md_content -> 图片替换(描述和地址) -> new_md_content

2、node 中的图片处理

2.1 节点作用

将 md 中的图片进行处理：

1
2
处理前： ![](images/f77da4df52517fc50b9efb528540e1351dd1a08dce6f801cf08366540f2c59ce.jpg)
处理后： ![图片的内容描述](http...)

这样处理有两方面优势：

  • 优势一：让图片真正参与到内容回答中（图片内容会被模型识别）。对应面试/追问点：你们的 RAG 是做的图文混合检索的么？
  • 优势二：本地地址换成网络地址，客户端在任何位置都可以访问和回显图片，回答问题的时候可以直接带上图片。

2.2 参数说明

  • 使用：md_path → md_content → images（图片文件夹）
  • 生成：md_content；文件由 文件名.md → 文件名_new.md

2.3 准备工作

  • 准备文件服务器 / 文件服务器的使用
    • 自己搭建：minio 搭建文件服务器（需要考虑服务器、搭建维护成本）
    • 直接买：阿里、腾讯、华为等 OSS 对象存储服务器（需要花钱）
  • 准备视觉模型对象
  • 在 infra 层封装：minio 的操作 | 模型的使用

2.4 实现步骤总览

  1. 安装了 minio 软件
  2. 准备 minio 的公共类（桶名、图片前缀、client、构建访问路径）
  3. 准备 llm 的公共类（获取大语言模型 / 获取视觉模型）
  4. 获取数据并且校验（state → md_path）→ md_content、images_path_obj、md_path_obj
  5. 判断 images 文件夹是否为空
  6. 为后续步骤做准备，提取图片的信息：每张图片 → (图片名.jpg, 图片地址, (上文, 下文))
  7. 使用视觉模型进行图片的意图识别
  8. 将图片传递到 minio，并且完成 md_content 内容的替换
  9. 替换 state 内容 md_content、md_path

3、step 4：参数获取与校验

获取 md_content、images_path_obj、md_path_obj 三个对象：

  1. 获取 md_path 属性（state）
  2. 非空和文件存在性校验
  3. 读取 md_content 内容（通过 md_path），md_content → state
    注意：md_content 必须存进 state，因为后面可能没有图片，后续步骤仍然要用到它
  4. 获取图片文件夹地址：md_path_obj → .parent / 'images' → images_path_obj
  5. 返回，供后续步骤使用

4、step 6：scan_images 提取图片上下文

图片文件夹与 md 内容的对应关系大致如下：

1
2
3
4
images 文件夹对象              md_content
    xxxxxx.jpg                   xxxxxxx ![](xxxxxx.jpg)  xxxxxxx
    xxxxxx.png                   jjjjjjj ![]( xxxxxx.png) jjjjjj
    .....

目标函数签名：

1
scan_images(images_path_obj: Path, md_content: str) -> list[tuple(图片名.xx, 图片地址, (上, 下))]

实现细节：

  1. 遍历图片文件夹 images，获取每一个文件
  2. 判断文件是不是图片 → 通过后缀判断，命中则加入 list[]
  3. 使用图片的名字 name → 在 md 中查找是否被引用（正则回顾，见第 7 节）
  4. 若被引用，获取引用信息 start、end
  5. 取上文 [start-100:start]、下文 [end:end+100]
  6. 拼接本次数据 (图片名, str(obj), (上文, 下文)) → append 到结果列表

最终产物：

1
2
md_content, images_path_obj, md_path_obj, image_context_list = ...
# image_context_list : list[tuple[str, str, tuple[str, str]]]

5、step 7：summarize_images 视觉模型识别

1
dict[str -> name, str -> 图片的描述] = summarize_images(image_context_list, 文件夹的名字 md_path_obj.stem)

实现细节：

  1. 获取模型对象 vision_chat()
  2. 封装提示词（图片 / 文本）
  3. 封装一个调用链 StrOutputParser()
  4. 执行调用链获取返回结果
  5. 封装一个 dict，并且返回即可

细节（面试/坑）：

  • base64
  • 视觉模型的数据格式
  • 滑动窗口限速器
  • langchain 的 chains 是 | 串联

6、step 8：上传 MinIO 并替换 md_content

1
2
3
md_content_new: str = upload_images_and_replace(
    md_content, summarize_images: dict, image_context_list, md_path_obj.stem
)

实现细节：

  1. 获取 minio 客户端对象
  2. 先删除当前文件图片（minio 上的）
  3. 循环 image_context_list 向 minio 上传文件 → dict[str: image_name, str: 网络地址]
  4. 循环”图片 → 网络地址”对应的字典，逐一替换 md_content 内容
  5. 返回 md_content_new

细节（面试/坑）：

  1. 查询 prefix 不能以 / 开头（处理方式 [1:]）
  2. 查询到的对象要转成 DeleteObject，即 [DeleteObject(o.object_name) ...]
  3. 删除 errors：for ... yield
  4. 上传用 fput(bucket_name, object_name, file_path, content_type=guess_type[0])
  5. 拼接访问地址
  6. 替换：md_content = re.sub(lambda _: f"![]()", md_content)

7、step 9：回写 state

  • 把 md_content 备份到原文件名对应的 文件名_new.md 文件中
  • 替换 state 中的 md_content、md_path

8、MinIO 的使用

8.1 搭建和安装（docker minio）

方案一：使用 docker —— 一行命令

1
2
3
4
5
6
docker run -d --name minio \
     -p 9000:9000 -p 9001:9001 \    # 端口映射：9000 -> api 接口端点，9001 -> 可视化管理页面
     -e "MINIO_ROOT_USER=minioadmin" \   # 账号
     -e "MINIO_ROOT_PASSWORD=minioadmin" \   # 密码
     -v $(pwd)/volumes/minio/data:/data \
     quay.io/minio/minio server /data --console-address ":9001"

注意：

  • docker 方式拉取的 minio 镜像是标准的社区版本；minio 企业版本需要认证（认证费）
  • 2025 年 5 月的时候，社区版本的功能被阉割了，去掉了权限修改的功能（只是阉割了 9001 的可视化管理端，使用代码都一样）

方案二：手工搭建

  • 这个版本提供了权限修改的功能

8.2 可视化操作和使用（minio 的核心概念）

  1. 访问 minio 客户端：ip : 9001
  2. 登录和认证：minioadmin / minioadmin
  3. 创建一个桶：一个项目对应一个桶
  4. 上传文件或者文件夹，桶内结构例如：
1
2
3
4
桶
  文件.png
  文件夹
     文件.png

  1. 访问地址规则

    • minio 部署的时候可以添加 ssl 证书（安全）
    • 地址格式：http | https → 端点 + 桶 + 桶内全路径（对象名）
    • 例如 http://ip:9000/0318tong/mv.jpg、http://ip:9000/0318tong/images/mv.jpg

  2. 拒绝访问问题

    • minio 文件服务器默认：上传需要认证（登录），访问也需要认证（直接访问会被拒绝）
    • 解决思路：1）访问时添加认证；2）修改桶权限——上传需要认证，但访问可以匿名

  3. 修改桶权限的三种模式

权限 上传 访问
private 需要认证 需要认证
public 不需要认证 不需要认证
custom 自定义 自定义

custom 自定义策略示例：

1
2
3
4
5
6
7
8
9
10
11
{
  "Version": "2012-10-17",            // 策略版本（固定值，兼容 S3 标准）
  "Statement": [
    {
      "Effect": "Allow",              // 策略效果：允许访问
      "Principal": {"AWS": ["*"]},    // 授权对象：所有用户
      "Action": ["s3:GetObject"],     // 授权操作：读取桶内对象
      "Resource": ["arn:aws:s3:::桶名/*"]   // 授权范围：桶内所有对象
    }
  ]
}

8.3 Python 中使用 minio 进行文件 CRUD

核心概念：

  • 端点 → ip地址:9000，具体的文件服务器地址
  • 桶 → 名称，主要是区分不同项目
  • 对象 → 具体存储的文件，如 xx.zip、xx/xxx/xxx.zip
  • 访问地址 → http | https:// + 端点 + 桶 + 对象

使用 Python 进行 minio 客户端登录：

1
2
3
4
5
6
7
8
from minio import Minio

client = Minio(
    "play.min.io",                  # 或 39.10.10.10:9000
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True | False,            # minio 是否使用加密证书，https | http
)

创建桶并设置权限：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
found = client.bucket_exists(bucket_name)
if not found:
    client.make_bucket(bucket_name)

    # Example anonymous read-only bucket policy.
    policy = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Effect": "Allow",
                "Principal": {"AWS": "*"},
                "Action": "s3:GetObject",          # 访问的权限
                "Resource": "arn:aws:s3:::桶名/*",
            },
        ],
    }
    client.set_bucket_policy("my-bucket", json.dumps(policy))

上传 / 删除 / 查看文件：

1
2
3
4
5
6
7
8
9
10
# 上传
put_object(bucket_name, object_name, 上传文件的字节数据, length, content_type="application/octet-stream")
fput_object(bucket_name, object_name, file_path, content_type="application/octet-stream")

# 删除
remove_objects(bucket_name, delete_object_list)

# 查看
get_object(bucket_name, object_name)                 # 查看单个文件
list_objects(bucket_name, prefix=None, recursive=False)   # 列出对象

list_objects 返回的层级大致是：

1
2
3
4
5
桶名
  文件夹名
     xx
     xx
     xx

8.4 封装一个 minio 使用的工具类

  • 专门创建客户端和返回客户端对象，隐藏桶的创建和权限的设置
  • 位置：app/shared/clients/minio_utils（读，不需要写）
  • 再封装一个公共对外访问类，位置：app/infra/object_stroage/minio_gateway（写）
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
from app.shared.clients.minio_utils import get_minio_client
from app.infra.config.providers import infra_config


class MinioGateway:

    # 提供获取桶名称的函数
    @property
    def bucket_name(self):
        return infra_config.minio_config.bucket_name

    # 提供获取图片前缀的函数
    @property
    def minio_img_dir(self):
        return infra_config.minio_config.minio_img_dir

    # 提供获取 minio_client 的函数
    @property
    def minio_client(self):
        return get_minio_client()

    # minio 上传文件不会返回访问地址 -> 我们拼接访问地址：http https 端点 桶 对象名
    # 封装一个拼接访问地址的函数
    def build_image_url(self, stem: str, object_name: str):
        """
         object_name = upload-images / 文件名 / 图片名称.png
        :param object_name:
        :return:
        """
        image_url = "https://" if infra_config.minio_config.minio_secure else "http://" + (
            f"{infra_config.minio_config.endpoint}"
            f"/{infra_config.minio_config.bucket_name}{infra_config.minio_config.minio_img_dir}/{stem}/{object_name}"
        )
        return image_url


minio_gateway = MinioGateway()

9、模型工具类和 infra 封装

9.1 解读模型初始化工具

模型初始化的公共工具（app/shared/model/lm_utils）负责按配置创建模型客户端，上层的 LLMProvider 只是对它做一层项目级封装。

9.2 封装一个模型对应的 infra

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
from app.infra.config.providers import infra_config
from app.shared.model.lm_utils import get_llm_client


class LLMProvider:
    # 获取普通大语言模型
    # def chat(self, json_mode: bool):
    #     """
    #       我们创建固定的语言模型 -> .env 中决定! 全局只有一种类型!
    #       我们可以在创建的时候传入 json 模式参数!!
    #     :param json_mode:
    #     :return:
    #     """
    def chat(self, mode_name: str, json_mode: bool):
        """
          我们允许传递模型的名称，不传入使用默认 .env > 默认参数
          我们可以在创建的时候传入 json 模式参数!!
          可以在一个项目使用不同的大语言模型
        :param json_mode:
        :return:
        """
        return get_llm_client(model=mode_name, json_mode=json_mode)

    # 获取视觉模型
    def vision_chat(self, vision_mode_name: str = None):
        return get_llm_client(vision_mode_name or infra_config.lm_config.lv_model)


llm_providers = LLMProvider()

10、正则使用回顾

10.1 基本用法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
import re

md_content = "这里有一张图 ![示意图](images/demo(1).png) 再来一张 ![图片](images/demo(1).png)"
image_name = "demo(1).png"

print(re.escape(image_name))

# 编译正则（不变） 正则中有两类字符：元字符 . * ? ! [a-z] {}，字面字符 a png
rep = re.compile(r"\!\[.*?\]\(.*?" + re.escape(image_name) + r".*?\)")

# 1. search：全文找【第一个】匹配 → 返回 Match 对象
match_obj = rep.search(md_content)
print("1. search 结果：", match_obj.group() if match_obj else "未找到")
print(f"1 获取匹配的位置 star {match_obj.start()} end {match_obj.end()} , span : {match_obj.span()}")

# 2. match：只从【开头】匹配 → 这里开头不是图片语法，必找不到
match_first = rep.match(md_content)
print("2. match 结果：", "找到" if match_first else "未找到（只匹配开头）")

# 3. findall：找【所有】匹配 → 返回字符串列表
all_list = rep.findall(md_content)
print("3. findall 列表：", all_list)

# 4. finditer：找【所有】匹配 → 返回 match 对象（相当于 search 匹配所有）
for m in rep.finditer(md_content):
    print("  - 内容：", m.group(), " | 位置：", m.span())

10.2 贪婪匹配与非贪婪匹配

1
2
3
4
5
6
7
text = "A123B456B"

# 1. 用 .*  → 贪婪（吃到最后）
print(re.findall(r"A(.*)B", text))    # 输出：123B456

# 2. 用 .*? → 非贪婪（找到第一个就停）
print(re.findall(r"A(.*?)B", text))   # 输出：123

要点小结：

  • re.escape() 用来转义图片名中的字面字符（如 demo(1).png 的括号），避免被当成元字符
  • search 找第一个、match 只从开头、findall 返回字符串列表、finditer 返回 Match 对象
  • Match 对象可通过 .group()、.start()、.end()、.span() 获取内容与位置
  • .* 贪婪、.*? 非贪婪，图片语法必须用非贪婪避免跨越多个图片