11.14 知识不止是文字 —— 多模态与垂直场景 RAG

痛点场景：你以为知识库都是干净的 Word 文档，现实却是：产品手册的关键步骤藏在截图里；财务知识的核心是一张张表格；客服质检要搜几万小时的通话录音；海外团队要用英文问中文制度库。到这一节，前面 13 节的“文本 RAG”装备已经全部就位，本节处理四类最常见的高难度场景：跨语言、表格问答、音视频 RAG、端侧部署。

---

痛点：文字只是知识的“冰山一角”

场景	知识的真实形态	纯文本 RAG 的下场
产品手册/图纸	截图、流程图、图文混排 PDF	图片直接被丢弃，答案漏掉关键步骤
财务/运营数据	多层表头、合并单元格的表格	切块把表格拆散，数字问答全错
客服/会议资产	通话录音、会议视频	压根没有文本层，无从入库
跨国团队	中文制度库，英文提问	英文问题检索不出中文答案

---

思路：多模态 RAG 的三条技术路线

处理“非文字”知识，业界有三条路线，按“信息离文字有多远”来选：

路线	怎么做	适合	代价
A. 文本化（先翻译成文字）	OCR / 图片描述（captioning）/ ASR 转写后走普通文本 RAG	信息主要承载在“内容”而非“版面”	有损：图表结构、版面含义丢失
B. 多模态嵌入（统一向量空间）	用 CLIP / ColPali 把图片、页面整体嵌入，与文本同库检索	版面、图表结构本身有语义（11.11 的 ColPali）	需要 GPU 索引多模态向量
C. 多模态大模型直读（VLM 摘要/作答）	将检索命中的页面原图直接输入多模态 LLM	复杂图表推理、截图问答	单次调用费用较高，上下文占用较大

实践中常用 A+C 组合：入库时为每张图或每页生成文本描述（路线 A，用于检索），命中后再把原图输入 VLM（路线 C，用于理解图表细节）。路线 B 采用 ColPali 一类多向量检索，更适合扫描件较多的资料库。

---

代码实现一：图文混排 PDF 的“描述 + 直读”组合拳

# 11.14 图文混排 RAG：图片文本化入库 + 命中原图给 VLM 精读
from pydantic import BaseModel, Field
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

vlm = ChatOpenAI(model="gpt-4o-mini", temperature=0, max_tokens=500)

def caption_image(image_b64: str) -> str:
    """路线 A：入库前给图片写'替身文本'。让图片内容也能被文本检索命中。"""
    msg = ChatPromptTemplate.from_messages([
        ("user", [
            {"type": "text", "text": "用 2~3 句中文描述这张图的内容，"
             "把图里的关键数字、步骤、结论都写出来。"},
            {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{image_b64}"}},
        ])
    ])
    return vlm.invoke(msg.format_messages()).content

# 入库：chunks += [Document(page_content=f"[图片] {caption_image(b64)}",
#                           metadata={"image_ref": "p7_img2.png", ...})]
# 检索：正常文本检索即可命中图片的替身文本
# 生成：命中后把 metadata 里 image_ref 指向的原图 + 问题一起交给 VLM 精读作答

---

代码实现二：跨语言 RAG —— 三种方案对号入座

“英文问题搜中文库”的本质是让问题和文档落进同一个语义空间：

方案	做法	适合	代表模型
多语言嵌入（首选）	嵌入模型天生跨语言，中英问题直接搜中文库	绝大多数场景	BGE-M3、multilingual-e5
查询翻译	先把问题翻译成文档语言再检索	嵌入模型跨语言能力弱时的补救	任意翻译模型/LLM
双语平行库	同一内容两个语言版本都入库	法务/合规要求“引用原文”的场景	解析管道加一道翻译

# 11.14 跨语言检索：多语言嵌入一库通吃
# 路线 A（默认，零下载）：直接用 .env 里的多语言 Embedding API 端点
# （方舟 doubao-embedding、OpenAI text-embedding-3 等）——见 code/shared_corpus.py
from shared_corpus import make_embeddings
import numpy as np

emb = make_embeddings()
doc_v = np.array(emb.embed_query("差旅报销单须在返回工作地后 5 个工作日内提交 OA 系统。"))
q_v = np.array(emb.embed_query("How many days to submit the expense report?"))
sim = doc_v @ q_v / (np.linalg.norm(doc_v) * np.linalg.norm(q_v))
# 实测英文问题 ≈ 0.56，中文问题 ≈ 0.70——跨语言命中足够排队第一

# 路线 B（本地部署，可选）：BGE-M3 约 2.3GB，数据不出门时用
# pip install sentence-transformers
# from sentence_transformers import SentenceTransformer
# emb = SentenceTransformer("BAAI/bge-m3")   # 100+ 语言对齐
# 或用 Ollama / Xinference 起本地 OpenAI 兼容 Embedding 服务，把
# .env 的 OPENAI_API_BASE 指过去，代码零改动。

⚠️ 跨语言的隐藏坑在生成层：检索命中了，但模型可能用英文回答中文文档。解法是在 11.12 的接地 Prompt 里加一条硬规则——“使用用户提问的语言回答，引用原文保持原文语言”。

---

代码实现三：音视频 RAG —— 时间戳就是你的元数据

音频/视频没有文本层，先转写、再切块、把时间戳写进元数据，检索结果就能直接跳转播放：

# 11.14 音视频 RAG：ASR 转写 → 时间窗切块 → 带时间戳元数据入库
# pip install faster-whisper
from faster_whisper import WhisperModel

def transcribe_with_timestamps(audio_path: str, window_sec: int = 45) -> list[dict]:
    model = WhisperModel("medium", compute_type="int8")
    segments, _ = model.transcribe(audio_path, language="zh", vad_filter=True)
    # 按固定时间窗聚合：45 秒一个 chunk，块与块留 5 秒重叠
    chunks, buf, start = [], [], None
    for seg in segments:
        if start is None:
            start = seg.start
        buf.append(seg.text)
        if seg.end - start >= window_sec:
            chunks.append({"start": start, "end": seg.end, "text": "".join(buf)})
            buf, start = [], seg.end - 5   # 5 秒重叠，防止话被拦腰切断
    if buf:
        chunks.append({"start": start, "end": seg.end, "text": "".join(buf)})
    return chunks

# 入库：Document(page_content=c["text"],
#                metadata={"media": "q3_review.mp4", "start": c["start"], "end": c["end"]})
# 前端：拿到命中 chunk 的 start/end，播放器直接 seek 过去——这就是“搜到哪句跳到哪句”

视频再加一路：用关键帧抽帧 + 上一节的图片描述，让“画面里的事”也可检索（比如“哪一页 PPT 讲了预算”）。ASR 引擎选型：本地/离线用 faster-whisper（Whisper 蒸馏加速版），云上可用各家语音转写 API，OpenAI Whisper 官方仓库见 github.com/openai/whisper。

配套脚本 s14_multimodal_rag.py 默认去读本地的 q3_review.mp3。仓库里没有塞录音——版权和体积都不划算——你自己放一段季度复盘、客服通话进去就能看到时间窗切块。文件不在时脚本会打印跳过，不会把整章演示卡死。

---

代码实现四：结构化数据 RAG —— 表格、CSV 与 JSON

痛点场景：同一批数据，放进 Word 里叫“文档”，放进 Excel 里叫“表”。散文句子讲的就是意思，塞进语义检索很合适；但把整张表切成 chunk 扔进向量库，等于让模型在一堆数字里做阅读理解——它检索到的是一堆碰巧相邻的单元格，既慢又容易把“3 月”和“毛利率”配错对。

一句话比喻：非结构化文档像散文，结构化数据像 Excel 表格。散文可以按段落切、按语义搜；表格是“行=一条记录、列=一个字段”的坐标网格，切碎了坐标就丢了。

三条路线：看你要“找一行”还是“算一个数”

路线	怎么做	适合	代价
① 把行变成句子	每行拼成“字段名: 值；字段名: 值”的一句话再入库，检索命中后连同原始行一起给模型	单点查找、字段问答	行数巨大时索引膨胀（上百万行要另想办法）
② schema 检索 + Text-to-SQL	表名、列名、字段说明也入库；先检索“哪张表、哪些列和问题相关”，再让模型写 SQL 交给数据库执行	聚合、比较、join	依赖 SQL 生成质量，必须有安全笼子（见下文 ## Text-to-SQL 必须运行在笼子里）
③ 混合	先用 ① 定位到相关行/表，再用 ② 在限定范围内生成受限 SQL 做聚合	既要找得到、又要算得对	链路最长，调试成本最高

💡 经验法则：“找一行”用路线 ①，“算一个数”用路线 ②。凡是问题里出现“一共/平均/最多/环比”这类词，答案必须由代码（SQL/pandas）算出来，LLM 只负责读懂意图和解读结果——理由与上文 ## Text-to-SQL 必须运行在笼子里 完全一致。

JSON：先展平，再把路径当元数据

嵌套 JSON（比如订单里套着商品数组）不要整块塞成一个 chunk。正确做法是展平：把 order.items[2].price 这样的字段路径拎出来，当成一条独立记录入库，同时把路径本身写进元数据。这样模型引用“第 2 个商品的价格”时，你能顺着路径回到原始 JSON 的那个字段去核对，而不是只给出一团 JSON 文本。

元数据必带：表名、主键、单位、口径、更新时间

这五项里，单位与口径缺失是表格问答最常见的翻车原因。金额是“元”还是“万元”、含不含税、是不是季度累计口径——不写清楚，模型算得再对也是错的。主键则是命根子：检索到一行后要能按主键回表，拿到这一行的最新值。

# 11.14 结构化数据 RAG：CSV 行 → 一句话 → 带表名/主键元数据入库
import csv
from langchain_core.documents import Document

def csv_to_docs(path: str, table: str, pk: str, unit_note: str = "") -> list[Document]:
    """把 CSV 每一行拼成“列名: 值”的一句话，元数据带上表名、主键与口径。"""
    docs = []
    with open(path, encoding="utf-8") as f:
        for row in csv.DictReader(f):                    # 每行是一条记录
            sentence = "；".join(f"{col}: {val}" for col, val in row.items() if val)
            docs.append(Document(                         # 行变句子，供语义检索命中
                page_content=f"[表 {table}] {sentence}",
                metadata={                               # 元数据是回表的坐标 + 口径说明
                    "table": table,                      # 哪个表
                    "primary_key": row[pk],              # 主键 → 按它回表拿最新值
                    "unit_note": unit_note,              # 金额单位：元，含税
                    "updated_at": "2026-09-01",          # 口径与时效
                    "columns": ",".join(row.keys()),      # 列名，供列级过滤
                },
            ))
    return docs

# 入库后：检索到句子 → 用 primary_key 回表 → 把整行原文交给模型；要聚合就转成 SQL

💡 什么时候升级到路线 ②？ 当问题开始出现“哪个月环比增长最快”“各部门平均报销多少”——句子检索给不出聚合结果，这时候就该把表名、列名和字段说明单独入库，走 schema 检索 + Text-to-SQL，并套上 ## Text-to-SQL 必须运行在笼子里 的六条护栏。

配套脚本：code/s14_multimodal_rag.py 的 rows_to_documents()（行 → 带表名/主键/单位元数据的句子文档）与 flatten_json()（嵌套 JSON 展平成字段路径 → 值）。

两份可直接跑的参考实现（同一作者的结构化 RAG 系列 notebook）：

• CSV RAG：把表格行转成可检索文本
• JSON RAG：嵌套结构展平后入库

---

拓展：表格问答与端侧 RAG 速览

表格问答（Table QA）

表格是结构化数据，不要用“切文本”的方式对待它（呼应 11.2 的入库策略）：

问题类型	推荐打法
“3 月的毛利率是多少？”（单点查找）	表格行级嵌入 + 表头语义增强，命中后整行返回
“哪个月环比增长最快？”（聚合/比较）	别让 LLM 硬算——把表格转成 SQL（Text-to-SQL）或 pandas，让代码执行，LLM 只负责“读懂意图和解读结果”
超宽表（上百列）	先做列筛选（按表头语义挑相关列），再进生成，避免上下文爆炸

💡 经验法则：数字计算一律下放给工具（SQL/pandas），LLM 只做它擅长的“理解与表达”。让大模型口算 23×47，是对它最残忍的 misuse。

端侧 RAG（跑在自己电脑/手机上）

在个人笔记、离线环境等数据需要留在本地的场景中，可以组合使用本地小模型：

组件	开箱即用方案
本地向量库	sqlite-vec（SQLite 扩展，零部署）
本地嵌入/生成模型	Ollama 一条命令拉模型，OpenAI 兼容接口
桌面知识库成品	AnythingLLM（11.10 选型地图里的“本地私有”选手）

端侧的代价是模型小、上下文短——切块要更小（11.2）、重排要更省（11.5）、生成层 Prompt 要更精简（11.12），本章所有“省着用”的技巧在端侧全部是刚需。

多模态评估不能只看“模型描述得像不像”

文本化、跨模态检索和 VLM 作答是三段不同能力，应分别测：

阶段	评测问题	典型指标/用例
OCR/ASR/图片描述	原始信息有没有被正确转成可检索内容	字错率、数字准确率、表格结构、说话人、时间戳
跨模态检索	文字问题能否找到正确页面/片段	Recall@K、MRR、页级命中率
VLM/工具作答	命中原图后能否正确读数与推理	答案正确性、引用页、坐标/时间定位

一张图描述得很流畅，关键数字却写错，仍然是失败。评测集应包含小字号、旋转扫描、跨页表格、相似图表、多人对话、口音、背景噪声和跨语言问题。

图文 PDF 要保留“页面坐标协议”

把图片描述存进向量库时，同时保存 page_id、原图路径、边界框、解析器版本和 OCR 置信度。命中后前端才能跳到正确页面并高亮区域。若只保存一句 caption，用户无法核查，重新解析后引用也可能漂移。

低 OCR 置信度的金额、日期和编号应进入人工复核或 VLM 二次读取，不能悄悄混进知识库。文本路线与视觉路线都命中同一页面时，可融合排名；两者结论冲突时应保留原图并提示复核。

音视频不只需要时间戳，还需要说话人和置信度

会议中“王经理说可以延期”和“客户不同意延期”含义相反。Chunk 元数据至少应保存开始/结束时间、说话人、ASR 置信度、媒体版本和访问权限。切窗要有重叠，但重叠片段进入上下文前需要去重。

视频可以把 ASR、关键帧描述、屏幕 OCR 和章节标题按时间轴对齐。检索命中后返回可播放区间，引用格式可写成 review.mp4#t=12:30-12:48。

Text-to-SQL 必须运行在笼子里

把计算交给 SQL/pandas比让 LLM 口算可靠，但工具调用增加了新的风险：

1. 数据库使用只读账号和允许表/列白名单；
2. 禁止 DDL/DML、多语句、外部函数和任意文件访问；
3. 设置查询超时、扫描行数、结果行数和费用预算；
4. 在执行前解析 SQL，而不是只用关键词正则；
5. 行级权限由服务端身份注入，模型不能自行选择租户；
6. 返回答案同时附查询、数据快照时间和结果来源。

配套脚本增加了只读操作门禁和独立的 ASR 时间窗函数，保留最低置信度，方便对低质量片段送审。

端侧能力也要按设备分层

“本地运行”不代表所有设备能力相同。桌面 GPU、普通笔记本和手机的内存、功耗、模型格式都不同。应记录冷启动、峰值内存、电量/温度、索引大小和离线更新时间；数据加密、系统备份和设备丢失后的远程擦除也属于端侧安全边界。

---

多模态资料要保留回到原件的路径

图片描述、OCR 文本和语音转写都是派生数据，必须能回到原页、原图或原时间段。保存页码、坐标、时间戳、说话人和置信度，用户才能核对模型引用。

表格问答要保留表头、单位和统计口径；音视频要测试专有名词与多人重叠说话；跨语言检索要分别评估召回与回答语言。不同材料的失败方式不同，不宜只用一个综合准确率概括。

---

权威官方参考

• BGE-M3：多语言多粒度嵌入（BAAI FlagEmbedding）
• ColPali：视觉文档检索（详见 11.11）
• faster-whisper：本地快速语音转写（SYSTRAN）
• OpenAI Whisper 官方仓库
• sqlite-vec：SQLite 向量检索扩展
• Ollama：本地大模型一键运行
• AnythingLLM：全栈本地 AI 知识库
• OpenCLIP：开源图文对齐模型实现
• RAG_Techniques：CSV 表格 RAG 实战 notebook
• RAG_Techniques：JSON 展平 RAG 实战 notebook