简介一份围绕 DeepSeek 多模态实战的 PDF 文档面向对文本生成与知识图谱构建感兴趣的 AI 工程师、数据开发者及进阶学习者。文档从实际应用出发系统讲解 DeepSeek 的基本原理、多模态处理能力并重点拆解“文本生成 知识图谱构建”协同 Pipeline 的设计思路覆盖数据预处理、实体识别、关系抽取、图谱存储与可视化、模型微调、文本生成实现、融合策略等关键环节适合用于快速建立完整技术框架。资源共 1 个 PDF 文件大小 2.02MB31 页内容完整、目录清晰所有文字和图表显示正常。文中还包含实战案例分析与常见问题解决方案能帮助读者规避数据噪声、实体对齐、生成质量不高等典型障碍。已有 107 人浏览学习适合作为从入门到进阶的实用参考资料。1. DeepSeek 多模态实战这份 PDF 讲的不只是调用 API做 DeepSeek 多模态项目的人大多卡在同一道坎接口调通了图片也能传上去但生成结果没法直接用——要么文本质量飘忽要么数据根本进不了知识图谱。这份 PDF 的标题很直白“文本生成 知识图谱构建的完整 Pipeline 设计”它把多模态输入怎么统一处理、动态文本生成怎么做、抽取出来的三元组怎么落库拆成了一条能照着搭的链路。适合正在做知识中台、RAG 增强、商品多模态支持或工业知识图谱的工程师也适合刚部署完 DeepSeek 但不知道怎么把业务串起来的团队。我按这份设计的思路复现了一遍下面把实现细节、参数和踩过的坑一条条写出来第一次跑通大概需要半天。2. 多模态统一处理与动态文本生成让模型先“看”到输入再组织语言2.1 多模态输入为什么不能当普通字符串拼进 prompt很多第一次接触 DeepSeek 多模态接口的人会犯一个错把图片路径写进文本比如“请分析 /data/img/001.jpg 这张图”然后发现模型回答“我无法访问该文件”。原因在于多模态模型不是通过文件系统读图的它的视觉编码器把图片切成 patch转成视觉 token再和文本 token 一起进 Transformer 上下文窗口。图像对模型来说是一种模态输入不是一串字符。这份 PDF 里强调的“多模态统一处理”本质上就是把图像、表格、音视频各自转成模型能消费的格式。图像走 base64 编码或公网 URL表格先文本化成 Markdown 或 CSV音视频抽帧抽轨后再描述。统一之后DeepSeek 才能在同一轮对话里同时理解“图里有什么”和“文本里写了什么”。多模态与纯文本调用还有一个关键差异视觉 token 数量大一张 1024x1024 的图可能要占几百个 token。上下文窗口是共享的图像越清晰、尺寸越大留给文本生成的空间就越小。所以统一处理里面必然要包括图像压缩这一步控制分辨率而不是原图直接怼上去。2.2 图像与表格的文本化预处理base64 与视觉编码器我按这份 PDF 里的做法实现了一个图像预处理函数先把图片转成 base64再通过 OpenAI 兼容接口传给 DeepSeek 的多模态模型。下面这段代码可以直接用于本地部署的 vllm 环境也可以用云端 APIimport base64 from openai import OpenAI def encode_image_to_base64(image_path: str) - str: 把本地图片编码为 base64 字符串接口要求 data URL 形式 with open(image_path, rb) as f: return base64.b64encode(f.read()).decode(utf-8) def describe_image(api_client: OpenAI, image_path: str) - str: 让多模态模型输出图片的纯文本描述供后续动态文本生成使用 b64 encode_image_to_base64(image_path) resp api_client.chat.completions.create( modeldeepseek-vl, messages[ { role: user, content: [ { type: image_url, image_url: {url: fdata:image/jpeg;base64,{b64}}, }, { type: text, text: 用不超过 50 个字描述这张图片中的核心物体、数量和工作状态只输出描述本身。, }, ], } ], temperature0.2, max_tokens128, ) return resp.choices[0].message.content.strip()逻辑说明第一段 encode_image_to_base64 处理的是本地文件vllm 部署时如果服务跑在另一台机器本地编码再传 data URL 是最稳妥的方式避免宿主机和容器文件系统不一致导致的路径玄学问题。第二段 describe_image 控制 temperature 在 0.2因为图像描述需要稳定不需要创意max_tokens 限制 128够描述核心信息又不至于浪费上下文。参数说明里最需要注意的是 temperature。很多人用文本生成的习惯把 temperature 设成 0.7 甚至 0.9放到图像描述任务上就会得到风格飘忽的输出同一个批次的描述格式五花八门后面做知识抽取时会非常痛苦。图像描述阶段的 temperature 我一般固定在 0.2 以内图片分辨率则在传入前用 Pillow 统一压到 768 像素短边兼顾细节和 token 占用。表格类的多模态输入不一定要走视觉接口。如果表格是图片格式可以参考上述思路先转成 Markdown如果源头就是 CSV 或 Excel直接读取转成文本更可控。多模态统一处理的价值就在这里把不同来源的输入归一化到同一种文本描述后面的动态文本生成才不用为每种输入单独写分支。2.3 动态文本生成用场景模板组合上下文拿到了图像描述下一步是生成真正的业务文本。动态文本生成不是简单把字符串拼起来而是按场景选择不同的提示词模板把视觉描述和结构化参数组合进上下文。以商品多模态支持为例商品主图描述和规格参数需要产出一条营销文案我的实现是SCENE_TEMPLATES { marketing: ( 你是电商运营专家。请根据【商品图描述】和【商品规格】生成一段 100 字以内的卖点文案。\n 要求突出 3 个核心卖点不用重复规格里的基础参数。\n 【商品图描述】{vision_note}\n 【商品规格】{specs} ), inspection: ( 你是设备维护工程师。请根据【现场图片描述】和【设备参数表】判断设备状态。\n 输出格式状态结论 依据。\n 【现场图片描述】{vision_note}\n 【设备参数表】{specs} ), } def generate_dynamic_text(api_client, scene, vision_note, specs): prompt SCENE_TEMPLATES[scene].format( vision_notevision_note, specsspecs ) resp api_client.chat.completions.create( modeldeepseek-vl, messages[{role: user, content: prompt}], temperature0.4, max_tokens512, ) return resp.choices[0].message.content.strip()逻辑说明模板把“视觉观察”和“结构化参数”分成两个占位符场景不同输出格式和约束也不同。营销场景要控制卖点数量巡检场景要保证输出格式稳定这些约束全部写进模板而不是靠调用方事后解析。temperature 在营销场景可以放到 0.4 给一点表达空间但别超过 0.5否则输出的信息密度会下降。动态的“动态”体现在模板选择可以依赖业务规则。比如商品属于数码类就选参数优先的模板属于服饰类就选视觉优先的模板设备告警级别高就切换成严格输出格式的巡检模板。这一步是做多模态项目最容易忽略的地方——模板不是写死一段 prompt而是把提示词本身变成可配置的模块。这也是这份 PDF 里 pipeline 设计的核心思路每个阶段的输出格式都要为下一个阶段服务文本生成的结果必须能被可靠地解析。社区里也有 Hermes 这类微调权重指令跟随能力在某些场景下更稳接口调用方式与 DeepSeek 一致。如果发现某个场景抽取或生成格式老是不稳定可以换这类微调版做对比改造代价很低模型名换一下就行。3. 文本生成结果再加工从自然语言里抽三元组喂给图谱3.1 为什么生成完还要再抽一遍动态文本生成产出的是一段自然语言但知识图谱构建要求结构化数据。图数据库里存的是节点和关系营销文案、巡检结论这种自由文本没法直接入库必须先抽成三元组也就是 头实体、关系、尾实体 这样一组结构。比如“空压机-01 发生轴承磨损故障”抽出来就是 (“空压机-01”, “发生故障”, “轴承磨损”)。这份 PDF 里把文本生成和知识抽取设计成两个独立阶段而不是让模型一次性输出最终结构化结果。原因很实际生成阶段如果强制输出 JSON文本质量和信息完整度会明显下降先自由生成、再抽三元组两个阶段各干各的活调参互不干扰。这属于先踩过坑再回头设计的思路直接一步到位求 JSON 的管线三元组召回率往往低得没法看。工业知识图谱场景里尤其明显。设备日志生成的环境描述、故障记录、维修工单天然是给“人读”的文本但你做故障归因、备件预测时需要的是 (设备, 关联备件, 备件型号) 这类结构化边。抽不好的话后面所有图谱查询都是白搭。3.2 提示词设计用 strict JSON 约束抽取输出我在抽取阶段用的方法也是 DeepSeek 文本接口但提示词和生成阶段完全不同。抽取要求的是稳定输出所以在系统消息里直接声明“仅返回 JSON 数组”并在示例里给出一条完整的三元组样本EXTRACT_PROMPT 你是知识抽取器。从下面的文本中抽取所有实体与关系输出 JSON 数组。 每条三元组的格式固定为 {head: ..., relation: ..., tail: ...}。 要求 1. 实体必须是文本中出现的原文不要改写。 2. relation 从候选关系词表里选[发生故障, 包含, 依赖, 关联备件, 位于]。 3. 如果无法确定关系丢弃该条候选不要编造。 4. 只输出 JSON 数组不要输出任何解释。 候选关系词表发生故障、包含、依赖、关联备件、位于 文本 {text} def extract_triples(api_client, text: str) - list[dict]: resp api_client.chat.completions.create( modeldeepseek-chat, messages[ {role: system, content: 你是一个严谨的知识抽取器只输出 JSON。}, {role: user, content: EXTRACT_PROMPT.format(texttext)}, ], temperature0.1, max_tokens1024, ) raw resp.choices[0].message.content.strip() return parse_triple_json(raw)逻辑说明候选关系词表的作用是约束关系类型的枚举值。工业场景下关系种类一般不超过几十个给模型一个白名单能大幅减少关系名混乱——比如“包含”和“包括”这种近义词白名单能强制收敛。要求实体必须是原文是为了后面做实体对齐时能匹配上如果模型自由改写实体图谱里会出现大量同一事物的不同写法。temperature 在这里设成 0.1比生成阶段更低。知识抽取本质是信息提取任务不是创造性任务温度越高JSON 格式越容易出问题关系选择越不稳定。max_tokens 给到 1024因为三元组数组的输出长度通常比单段文本长。如果文本较长建议先按段落切分再逐段抽取不然输出超过 max_tokens 会被截断。3.3 解析与兜底JSON 坏了不等于结果丢了即便提示词约束得再死模型偶尔也会在 JSON 外包一层解释文字或者在某一行多打一个逗号。直接 json.loads 失败就丢弃整批结果是最亏的翻车方式。我一般会做两层兜底import json, re def parse_triple_json(raw: str) - list[dict]: 优先严格解析失败后用正则兜底提取三元组 try: data json.loads(raw) if isinstance(data, list): return [item for item in data if all(k in item for k in (head, relation, tail))] except json.JSONDecodeError: pass triples [] pattern re.compile(r\{head:\s*(.*?),\s*relation:\s*(.*?),\s*tail:\s*(.*?)\}) for match in pattern.finditer(raw): triples.append({head: match.group(1), relation: match.group(2), tail: match.group(3)}) return triples逻辑说明先尝试标准 JSON 解析成功率最高失败后不是重试模型调用而是用正则抓取所有三元组片段。正则兜底能挽回一部分结果但只处理简单格式如果模型把 JSON 结构破坏得太厉害兜底也拿不回来这时候才需要重试一次接口调用。有一个细节容易踩模型可能在 JSON 数组外层加了“json”标记。这种 Markdown 代码块包裹在 strict 模式下偶尔会出现解析前先做一个清洗把首尾的json 和 剥掉。我自己习惯在 parse_triple_json 里加一步替代清洗省得每次调用前还要单独预处理。3.4 用视觉实体校验文本抽取这是我在这份 PDF 基础上自己加的一步多模态能力的二次利用。文本抽取出的实体可以回过去问一次视觉模型确认该实体在图像里是否可见。比如文本里抽出了“轴承磨损”但巡检图片里根本没有轴承部件那这条三元组可能是上下文引用来的不一定是当前设备的真实状态。视觉校验的典型做法是把抽取出来的实体列表拼进提示词让多模态模型返回每个实体是否可见的布尔数组。这一步独立于图谱入库但能显著提升工业场景下三元组的可信度。多模态融合算法在这里的价值不在于识别得多准而在于给文本抽取加了一个视觉约束两个模态相互印证。校验通过的三元组才允许入库校验失败的单独存到疑点表后续人工复核。4. 知识图谱构建与存储从三元组到 Neo4j 可查询图4.1 图模型怎么设计节点、关系、属性的粒度三元组只是中间形态真正落到知识图谱存储时要先设计图模型。以工业知识图谱为例我会把节点分成四类Device设备、Fault故障、Part备件、WorkOrder工单。关系类型直接复用抽取阶段的白名单设备到故障是“发生故障”设备到备件是“关联备件”故障到备件是“依赖”。设计粒度上有一个原则能合并的节点不要拆分。比如“轴承磨损”和“轴承裂纹”如果后续分析只关心故障大类拆成两个独立节点会让图谱稀疏化查询时 count 统计被分散如果后续要做根因分析就必须拆开。可以先按大类入库后面发现查询需求细化再加标签Neo4j 支持动态加属性不需要重建图。属性设计方面我的习惯是给每个节点加 source 和 created_at 两个通用属性source 记录这条三元组来自哪份文档或哪张图created_at 记录入库时间。不要小看这两个字段图谱运行一段时间后排查数据质量问题时全靠它们定位来源。4.2 Cypher 导入唯一约束与 MERGE 的正确姿势导入 Neo4j 的第一步不是写 MERGE而是建唯一约束。没有唯一约束时MERGE 的判重逻辑会退化同一条数据可能生成多个重复节点。下面是建约束和批量导入的 CypherCREATE CONSTRAINT device_name IF NOT EXISTS FOR (d:Device) REQUIRE d.name IS UNIQUE; CREATE CONSTRAINT fault_code IF NOT EXISTS FOR (f:Fault) REQUIRE f.code IS UNIQUE;UNWIND $rows AS row MERGE (d:Device {name: row.device}) MERGE (f:Fault {code: row.fault_code}) MERGE (d)-[:HAS_FAULT {occurred_at: row.time, source: row.source}]-(f)逻辑说明UNWIND 把参数数组里的每行展开MERGE 先按唯一键找节点找到就复用找不到才创建。关系上的 occurred_at 和 source 是边的属性记录了故障发生时间和来源查询时间窗口时直接过滤边属性比过滤节点属性更快。参数说明$rows 是一个 JSON 数组每行至少包含 device、fault_code、time、source 四个字段。分批导入时我一般每批 500 到 1000 行太大一个事务容易内存溢出太小又浪费网络往返。Cypher 事务里如果某一行报错整个批次会回滚所以数据入库前最好先做一次字段完整性检查空 device 名、空 relation 都会坑掉整批数据。另一个经验关系类型名尽量用大写字母和下划线不要带空格。Neo4j 的关系类型是固定枚举改成带空格的名字后期写查询要加反引号麻烦且容易错。4.3 业务查询模板从图谱拿结论图谱建完不是终点业务查询才是。下面这条 Cypher 查的是“某台设备最近三个月的高频故障 Top 10”在设备运维场景里很常用MATCH (d:Device)-[r:HAS_FAULT]-(f:Fault) WHERE d.name 空压机-01 AND r.occurred_at datetime(2025-06-01T00:00:00) RETURN f.code, count(r) AS fault_cnt ORDER BY fault_cnt DESC LIMIT 10;逻辑说明WHERE 里同时过滤了节点属性 d.name 和边属性 r.occurred_at边属性过滤会走关系扫描数据量大的时候建议在 occurred_at 上建索引。count(r) 统计的是关系数量而不是节点数量因为同一个故障可能在多台设备上重复发生。查询性能方面设备数超过十万级的图谱建议给 Device.name 和 Fault.code 建索引关系属性索引只能覆盖边上的过滤条件。工业知识图谱绝大多数查询都固定在设备维度按设备名出发做图遍历索引覆盖住就能满足秒级响应。5. 完整 Pipeline 整合与避坑串起来之前先看这四条5.1 编排脚本把前面四个阶段合成一条命令前面几章的逻辑是分开写的实际落地时要用一条命令把图像描述、动态文本生成、三元组抽取、图谱入库串起来。我会用一个 Python 脚本管理整条 pipeline并且给每个阶段做断点缓存。这样任一步骤失败修复后不用从头跑一遍直接跳过已有结果继续import json, hashlib from pathlib import Path def md5_of_text(text: str) - str: return hashlib.md5(text.encode(utf-8)).hexdigest() def run_pipeline(api_client, data_dir: Path, cache_dir: Path): cache_dir.mkdir(exist_okTrue) for image_path in sorted(data_dir.glob(*.jpg)): img_key md5_of_text(image_path.name) stage1_file cache_dir / f{img_key}_desc.json stage2_file cache_dir / f{img_key}_text.json stage3_file cache_dir / f{img_key}_triples.json if not stage1_file.exists(): vision_note describe_image(api_client, str(image_path)) stage1_file.write_text(json.dumps({vision_note: vision_note}, ensure_asciiFalse)) if not stage2_file.exists(): vision_note json.loads(stage1_file.read_text())[vision_note] specs load_specs_for_image(image_path) text generate_dynamic_text(api_client, inspection, vision_note, specs) stage2_file.write_text(json.dumps({text: text}, ensure_asciiFalse)) if not stage3_file.exists(): text json.loads(stage2_file.read_text())[text] triples extract_triples(api_client, text) stage3_file.write_text(json.dumps(triples, ensure_asciiFalse)) write_triples_to_neo4j(json.loads(stage3_file.read_text()), sourceimage_path.name)逻辑说明每个阶段先检查缓存文件是否存在存在就直接读不存在才调接口。这样批量处理几百张图片时中间接口抖一下、网络抖一下重新执行脚本只会从失败的那个文件继续。断点是整个 pipeline 设计里被低估的一环没有断点的脚本跑半小时挂在中间只能从头再来这种血泪教训一次就够。参数说明缓存文件的命名用的是图片名 MD5避免文件名带中文或特殊字符引发路径问题。load_specs_for_image 是业务侧函数根据图片名从参数表里找对应的结构化数据这里省略了实现。write_triples_to_neo4j 内部用上一章的 Cypher 模板统一处理导入逻辑。5.2 避坑记录一图谱里出现了大量重复节点现象跑完 pipeline 后查 count发现“空压机-01”节点出现十几次数据量虚高查询结果全乱。原因抽取阶段模型输出了“空压机-01”和“空压机 - 01”两种写法实体规范化没做。MERGE 判重基于 name 精确匹配空格差异直接绕过判重。解决入库前增加实体规范化函数去掉所有空白字符、全角转半角、统一大小写。我在规范化里还加了一个同义词映射表把“空压机1号”“空压机 1 号”这类写法统一归一到“空压机-01”。从那以后图谱节点数量稳定在预期范围不再虚增。5.3 避坑记录二批量调用 DeepSeek 时频繁超时现象单张图片调用正常批处理跑 200 张时每隔几分钟就抛一次超时异常中断整个 pipeline。原因循环里同步调用接口上一请求还没返回就发下一个服务端并发压力大接口响应时间被拉长本地网络环境也不是专线偶发丢包会被超时放大。解决限制并发数用信号量控制同时进行中的请求不超过 5 个超时时间从默认 60 秒调到 120 秒增加指数退避重试第一次失败等 2 秒重试第二次等 4 秒。改完后批处理稳定跑完超时率从百分之十几降到接近零。5.4 避坑记录三三元组解析失败导致入库条数大幅缩水现象某批次 500 条文本入库的三元组只有 200 多条大量候选关系被丢弃。原因抽取提示词里没限制“不要输出解释”模型在 JSON 数组前后加了“以下是抽取结果”这类话严格 json.loads 失败后正则兜底无法匹配这些额外内容整段丢弃。解决在系统消息里加一句“只输出 JSON 数组任何解释都会被视为错误”并把温度从 0.1 下调到 0同时把正则兜底的模式改成允许前后存在任意非 JSON 字符。处理后该批次三元组数量回升到 450 条以上。5.5 避坑记录四Neo4j 批量导入时事务回滚现象一次导入 3000 行时报内存不足整个事务回滚中间结果全部丢失。原因UNWIND 一次展开 3000 行MERGE 创建的节点和关系对象全在事务内存里内存占用超限触发回滚。解决把导入改成每 500 行提交一次事务用循环分批。Cypher 本身不支持事务控制但事务边界由驱动 API 控制Python 驱动里每 500 行执行一次 session.execute_write 就是一个干净事务。批量导入不是一个事务越大越快合理批次反而更稳。5.6 边界条件什么场景不建议用这套 pipeline这套链路不是万能的。如果业务只有纯文本、没有图片多模态需求直接调 DeepSeek 文本接口做抽取就行引入图像描述阶段只会增加 latency 和成本。实体关系高度不固定的场景比如开放域问答里的即兴关系也不适合知识图谱——关系类型不收敛图谱会退化成一个大号 JSON 存储查询收益为零。成本也是边界之一。每张图至少产生两次模型调用图像描述一次、文本生成一次抽取可能还要第三次数据量上百万时这个成本必须提前评估。PDF 里给的 pipeline 设计更适用于数据万级到十万级的中型业务再大的规模要考虑离线处理或消息队列而不是单机脚本硬跑。6. 质量验证与回捞上线前先跑一遍覆盖率检查6.1 覆盖率与孤立节点检查脚本图谱导入完成后先别急着接业务查询。我习惯先跑一遍质量检查看看数据像不像样。检查脚本就几条简单 Cypher 统计def verify_graph(session): total_nodes session.run(MATCH (n) RETURN count(n) AS c).single()[c] orphan_nodes session.run( MATCH (n) WHERE NOT (n)--() RETURN count(n) AS c ).single()[c] relation_types session.run( MATCH ()-[r]-() RETURN type(r) AS t, count(r) AS c ).data() print(f总节点数: {total_nodes}, 孤立节点: {orphan_nodes}, 占比: {orphan_nodes / total_nodes:.2%}) for item in relation_types: print(f关系 {item[t]}: {item[c]})逻辑说明孤立节点占比是最直观的体检指标。如果超过 10%大概率是抽取阶段的关系类型没匹配上白名单导致边没建起来。关系类型分布能看出数据是否偏科——比如“关联备件”只有几十条而“发生故障”有几万条说明备件数据来源缺失后续要补采集。覆盖率检查通过后我才会把查询模板接到 Dashboard 上。6.2 回捞机制低置信度关系重新生成检查中发现某个关系类型数量异常少不用整批重新跑 pipeline。我的做法是定位到具体来源把对应文本单独捞出来换一个更低温度、更严格约束的提示词重新抽取只更新差异部分。回捞是增量操作涉及的数据量通常只有几百条成本可控。从那以后我每次跑完 pipeline 都强制走一遍覆盖率检查孤立节点比例超过阈值就直接回捞不再抱着侥幸心理往下走。质量检查这一关设好后面接 RAG 也好、接智能问答也好返回结果才不会时灵时不灵。希望这份从多模态输入到图谱入库的完整实现能帮你少烧几个晚上调试 prompt。本文还有配套的精品资源点击获取 
长河君 · 2026/10/25 14:27:21 · 阅读 1.2k DeepSeek多模态实战：从图像描述到知识图谱的完整Pipeline
简介一份围绕 DeepSeek 多模态实战的 PDF 文档面向对文本生成与知识图谱构建感兴趣的 AI 工程师、数据开发者及进阶学习者。文档从实际应用出发系统讲解 DeepSeek 的基本原理、多模态处理能力并重点拆解“文本生成 知识图谱构建”协同 Pipeline 的设计思路覆盖数据预处理、实体识别、关系抽取、图谱存储与可视化、模型微调、文本生成实现、融合策略等关键环节适合用于快速建立完整技术框架。资源共 1 个 PDF 文件大小 2.02MB31 页内容完整、目录清晰所有文字和图表显示正常。文中还包含实战案例分析与常见问题解决方案能帮助读者规避数据噪声、实体对齐、生成质量不高等典型障碍。已有 107 人浏览学习适合作为从入门到进阶的实用参考资料。1. DeepSeek 多模态实战这份 PDF 讲的不只是调用 API做 DeepSeek 多模态项目的人大多卡在同一道坎接口调通了图片也能传上去但生成结果没法直接用——要么文本质量飘忽要么数据根本进不了知识图谱。这份 PDF 的标题很直白“文本生成 知识图谱构建的完整 Pipeline 设计”它把多模态输入怎么统一处理、动态文本生成怎么做、抽取出来的三元组怎么落库拆成了一条能照着搭的链路。适合正在做知识中台、RAG 增强、商品多模态支持或工业知识图谱的工程师也适合刚部署完 DeepSeek 但不知道怎么把业务串起来的团队。我按这份设计的思路复现了一遍下面把实现细节、参数和踩过的坑一条条写出来第一次跑通大概需要半天。2. 多模态统一处理与动态文本生成让模型先“看”到输入再组织语言2.1 多模态输入为什么不能当普通字符串拼进 prompt很多第一次接触 DeepSeek 多模态接口的人会犯一个错把图片路径写进文本比如“请分析 /data/img/001.jpg 这张图”然后发现模型回答“我无法访问该文件”。原因在于多模态模型不是通过文件系统读图的它的视觉编码器把图片切成 patch转成视觉 token再和文本 token 一起进 Transformer 上下文窗口。图像对模型来说是一种模态输入不是一串字符。这份 PDF 里强调的“多模态统一处理”本质上就是把图像、表格、音视频各自转成模型能消费的格式。图像走 base64 编码或公网 URL表格先文本化成 Markdown 或 CSV音视频抽帧抽轨后再描述。统一之后DeepSeek 才能在同一轮对话里同时理解“图里有什么”和“文本里写了什么”。多模态与纯文本调用还有一个关键差异视觉 token 数量大一张 1024x1024 的图可能要占几百个 token。上下文窗口是共享的图像越清晰、尺寸越大留给文本生成的空间就越小。所以统一处理里面必然要包括图像压缩这一步控制分辨率而不是原图直接怼上去。2.2 图像与表格的文本化预处理base64 与视觉编码器我按这份 PDF 里的做法实现了一个图像预处理函数先把图片转成 base64再通过 OpenAI 兼容接口传给 DeepSeek 的多模态模型。下面这段代码可以直接用于本地部署的 vllm 环境也可以用云端 APIimport base64 from openai import OpenAI def encode_image_to_base64(image_path: str) - str: 把本地图片编码为 base64 字符串接口要求 data URL 形式 with open(image_path, rb) as f: return base64.b64encode(f.read()).decode(utf-8) def describe_image(api_client: OpenAI, image_path: str) - str: 让多模态模型输出图片的纯文本描述供后续动态文本生成使用 b64 encode_image_to_base64(image_path) resp api_client.chat.completions.create( modeldeepseek-vl, messages[ { role: user, content: [ { type: image_url, image_url: {url: fdata:image/jpeg;base64,{b64}}, }, { type: text, text: 用不超过 50 个字描述这张图片中的核心物体、数量和工作状态只输出描述本身。, }, ], } ], temperature0.2, max_tokens128, ) return resp.choices[0].message.content.strip()逻辑说明第一段 encode_image_to_base64 处理的是本地文件vllm 部署时如果服务跑在另一台机器本地编码再传 data URL 是最稳妥的方式避免宿主机和容器文件系统不一致导致的路径玄学问题。第二段 describe_image 控制 temperature 在 0.2因为图像描述需要稳定不需要创意max_tokens 限制 128够描述核心信息又不至于浪费上下文。参数说明里最需要注意的是 temperature。很多人用文本生成的习惯把 temperature 设成 0.7 甚至 0.9放到图像描述任务上就会得到风格飘忽的输出同一个批次的描述格式五花八门后面做知识抽取时会非常痛苦。图像描述阶段的 temperature 我一般固定在 0.2 以内图片分辨率则在传入前用 Pillow 统一压到 768 像素短边兼顾细节和 token 占用。表格类的多模态输入不一定要走视觉接口。如果表格是图片格式可以参考上述思路先转成 Markdown如果源头就是 CSV 或 Excel直接读取转成文本更可控。多模态统一处理的价值就在这里把不同来源的输入归一化到同一种文本描述后面的动态文本生成才不用为每种输入单独写分支。2.3 动态文本生成用场景模板组合上下文拿到了图像描述下一步是生成真正的业务文本。动态文本生成不是简单把字符串拼起来而是按场景选择不同的提示词模板把视觉描述和结构化参数组合进上下文。以商品多模态支持为例商品主图描述和规格参数需要产出一条营销文案我的实现是SCENE_TEMPLATES { marketing: ( 你是电商运营专家。请根据【商品图描述】和【商品规格】生成一段 100 字以内的卖点文案。\n 要求突出 3 个核心卖点不用重复规格里的基础参数。\n 【商品图描述】{vision_note}\n 【商品规格】{specs} ), inspection: ( 你是设备维护工程师。请根据【现场图片描述】和【设备参数表】判断设备状态。\n 输出格式状态结论 依据。\n 【现场图片描述】{vision_note}\n 【设备参数表】{specs} ), } def generate_dynamic_text(api_client, scene, vision_note, specs): prompt SCENE_TEMPLATES[scene].format( vision_notevision_note, specsspecs ) resp api_client.chat.completions.create( modeldeepseek-vl, messages[{role: user, content: prompt}], temperature0.4, max_tokens512, ) return resp.choices[0].message.content.strip()逻辑说明模板把“视觉观察”和“结构化参数”分成两个占位符场景不同输出格式和约束也不同。营销场景要控制卖点数量巡检场景要保证输出格式稳定这些约束全部写进模板而不是靠调用方事后解析。temperature 在营销场景可以放到 0.4 给一点表达空间但别超过 0.5否则输出的信息密度会下降。动态的“动态”体现在模板选择可以依赖业务规则。比如商品属于数码类就选参数优先的模板属于服饰类就选视觉优先的模板设备告警级别高就切换成严格输出格式的巡检模板。这一步是做多模态项目最容易忽略的地方——模板不是写死一段 prompt而是把提示词本身变成可配置的模块。这也是这份 PDF 里 pipeline 设计的核心思路每个阶段的输出格式都要为下一个阶段服务文本生成的结果必须能被可靠地解析。社区里也有 Hermes 这类微调权重指令跟随能力在某些场景下更稳接口调用方式与 DeepSeek 一致。如果发现某个场景抽取或生成格式老是不稳定可以换这类微调版做对比改造代价很低模型名换一下就行。3. 文本生成结果再加工从自然语言里抽三元组喂给图谱3.1 为什么生成完还要再抽一遍动态文本生成产出的是一段自然语言但知识图谱构建要求结构化数据。图数据库里存的是节点和关系营销文案、巡检结论这种自由文本没法直接入库必须先抽成三元组也就是 头实体、关系、尾实体 这样一组结构。比如“空压机-01 发生轴承磨损故障”抽出来就是 (“空压机-01”, “发生故障”, “轴承磨损”)。这份 PDF 里把文本生成和知识抽取设计成两个独立阶段而不是让模型一次性输出最终结构化结果。原因很实际生成阶段如果强制输出 JSON文本质量和信息完整度会明显下降先自由生成、再抽三元组两个阶段各干各的活调参互不干扰。这属于先踩过坑再回头设计的思路直接一步到位求 JSON 的管线三元组召回率往往低得没法看。工业知识图谱场景里尤其明显。设备日志生成的环境描述、故障记录、维修工单天然是给“人读”的文本但你做故障归因、备件预测时需要的是 (设备, 关联备件, 备件型号) 这类结构化边。抽不好的话后面所有图谱查询都是白搭。3.2 提示词设计用 strict JSON 约束抽取输出我在抽取阶段用的方法也是 DeepSeek 文本接口但提示词和生成阶段完全不同。抽取要求的是稳定输出所以在系统消息里直接声明“仅返回 JSON 数组”并在示例里给出一条完整的三元组样本EXTRACT_PROMPT 你是知识抽取器。从下面的文本中抽取所有实体与关系输出 JSON 数组。 每条三元组的格式固定为 {head: ..., relation: ..., tail: ...}。 要求 1. 实体必须是文本中出现的原文不要改写。 2. relation 从候选关系词表里选[发生故障, 包含, 依赖, 关联备件, 位于]。 3. 如果无法确定关系丢弃该条候选不要编造。 4. 只输出 JSON 数组不要输出任何解释。 候选关系词表发生故障、包含、依赖、关联备件、位于 文本 {text} def extract_triples(api_client, text: str) - list[dict]: resp api_client.chat.completions.create( modeldeepseek-chat, messages[ {role: system, content: 你是一个严谨的知识抽取器只输出 JSON。}, {role: user, content: EXTRACT_PROMPT.format(texttext)}, ], temperature0.1, max_tokens1024, ) raw resp.choices[0].message.content.strip() return parse_triple_json(raw)逻辑说明候选关系词表的作用是约束关系类型的枚举值。工业场景下关系种类一般不超过几十个给模型一个白名单能大幅减少关系名混乱——比如“包含”和“包括”这种近义词白名单能强制收敛。要求实体必须是原文是为了后面做实体对齐时能匹配上如果模型自由改写实体图谱里会出现大量同一事物的不同写法。temperature 在这里设成 0.1比生成阶段更低。知识抽取本质是信息提取任务不是创造性任务温度越高JSON 格式越容易出问题关系选择越不稳定。max_tokens 给到 1024因为三元组数组的输出长度通常比单段文本长。如果文本较长建议先按段落切分再逐段抽取不然输出超过 max_tokens 会被截断。3.3 解析与兜底JSON 坏了不等于结果丢了即便提示词约束得再死模型偶尔也会在 JSON 外包一层解释文字或者在某一行多打一个逗号。直接 json.loads 失败就丢弃整批结果是最亏的翻车方式。我一般会做两层兜底import json, re def parse_triple_json(raw: str) - list[dict]: 优先严格解析失败后用正则兜底提取三元组 try: data json.loads(raw) if isinstance(data, list): return [item for item in data if all(k in item for k in (head, relation, tail))] except json.JSONDecodeError: pass triples [] pattern re.compile(r\{head:\s*(.*?),\s*relation:\s*(.*?),\s*tail:\s*(.*?)\}) for match in pattern.finditer(raw): triples.append({head: match.group(1), relation: match.group(2), tail: match.group(3)}) return triples逻辑说明先尝试标准 JSON 解析成功率最高失败后不是重试模型调用而是用正则抓取所有三元组片段。正则兜底能挽回一部分结果但只处理简单格式如果模型把 JSON 结构破坏得太厉害兜底也拿不回来这时候才需要重试一次接口调用。有一个细节容易踩模型可能在 JSON 数组外层加了“json”标记。这种 Markdown 代码块包裹在 strict 模式下偶尔会出现解析前先做一个清洗把首尾的json 和 剥掉。我自己习惯在 parse_triple_json 里加一步替代清洗省得每次调用前还要单独预处理。3.4 用视觉实体校验文本抽取这是我在这份 PDF 基础上自己加的一步多模态能力的二次利用。文本抽取出的实体可以回过去问一次视觉模型确认该实体在图像里是否可见。比如文本里抽出了“轴承磨损”但巡检图片里根本没有轴承部件那这条三元组可能是上下文引用来的不一定是当前设备的真实状态。视觉校验的典型做法是把抽取出来的实体列表拼进提示词让多模态模型返回每个实体是否可见的布尔数组。这一步独立于图谱入库但能显著提升工业场景下三元组的可信度。多模态融合算法在这里的价值不在于识别得多准而在于给文本抽取加了一个视觉约束两个模态相互印证。校验通过的三元组才允许入库校验失败的单独存到疑点表后续人工复核。4. 知识图谱构建与存储从三元组到 Neo4j 可查询图4.1 图模型怎么设计节点、关系、属性的粒度三元组只是中间形态真正落到知识图谱存储时要先设计图模型。以工业知识图谱为例我会把节点分成四类Device设备、Fault故障、Part备件、WorkOrder工单。关系类型直接复用抽取阶段的白名单设备到故障是“发生故障”设备到备件是“关联备件”故障到备件是“依赖”。设计粒度上有一个原则能合并的节点不要拆分。比如“轴承磨损”和“轴承裂纹”如果后续分析只关心故障大类拆成两个独立节点会让图谱稀疏化查询时 count 统计被分散如果后续要做根因分析就必须拆开。可以先按大类入库后面发现查询需求细化再加标签Neo4j 支持动态加属性不需要重建图。属性设计方面我的习惯是给每个节点加 source 和 created_at 两个通用属性source 记录这条三元组来自哪份文档或哪张图created_at 记录入库时间。不要小看这两个字段图谱运行一段时间后排查数据质量问题时全靠它们定位来源。4.2 Cypher 导入唯一约束与 MERGE 的正确姿势导入 Neo4j 的第一步不是写 MERGE而是建唯一约束。没有唯一约束时MERGE 的判重逻辑会退化同一条数据可能生成多个重复节点。下面是建约束和批量导入的 CypherCREATE CONSTRAINT device_name IF NOT EXISTS FOR (d:Device) REQUIRE d.name IS UNIQUE; CREATE CONSTRAINT fault_code IF NOT EXISTS FOR (f:Fault) REQUIRE f.code IS UNIQUE;UNWIND $rows AS row MERGE (d:Device {name: row.device}) MERGE (f:Fault {code: row.fault_code}) MERGE (d)-[:HAS_FAULT {occurred_at: row.time, source: row.source}]-(f)逻辑说明UNWIND 把参数数组里的每行展开MERGE 先按唯一键找节点找到就复用找不到才创建。关系上的 occurred_at 和 source 是边的属性记录了故障发生时间和来源查询时间窗口时直接过滤边属性比过滤节点属性更快。参数说明$rows 是一个 JSON 数组每行至少包含 device、fault_code、time、source 四个字段。分批导入时我一般每批 500 到 1000 行太大一个事务容易内存溢出太小又浪费网络往返。Cypher 事务里如果某一行报错整个批次会回滚所以数据入库前最好先做一次字段完整性检查空 device 名、空 relation 都会坑掉整批数据。另一个经验关系类型名尽量用大写字母和下划线不要带空格。Neo4j 的关系类型是固定枚举改成带空格的名字后期写查询要加反引号麻烦且容易错。4.3 业务查询模板从图谱拿结论图谱建完不是终点业务查询才是。下面这条 Cypher 查的是“某台设备最近三个月的高频故障 Top 10”在设备运维场景里很常用MATCH (d:Device)-[r:HAS_FAULT]-(f:Fault) WHERE d.name 空压机-01 AND r.occurred_at datetime(2025-06-01T00:00:00) RETURN f.code, count(r) AS fault_cnt ORDER BY fault_cnt DESC LIMIT 10;逻辑说明WHERE 里同时过滤了节点属性 d.name 和边属性 r.occurred_at边属性过滤会走关系扫描数据量大的时候建议在 occurred_at 上建索引。count(r) 统计的是关系数量而不是节点数量因为同一个故障可能在多台设备上重复发生。查询性能方面设备数超过十万级的图谱建议给 Device.name 和 Fault.code 建索引关系属性索引只能覆盖边上的过滤条件。工业知识图谱绝大多数查询都固定在设备维度按设备名出发做图遍历索引覆盖住就能满足秒级响应。5. 完整 Pipeline 整合与避坑串起来之前先看这四条5.1 编排脚本把前面四个阶段合成一条命令前面几章的逻辑是分开写的实际落地时要用一条命令把图像描述、动态文本生成、三元组抽取、图谱入库串起来。我会用一个 Python 脚本管理整条 pipeline并且给每个阶段做断点缓存。这样任一步骤失败修复后不用从头跑一遍直接跳过已有结果继续import json, hashlib from pathlib import Path def md5_of_text(text: str) - str: return hashlib.md5(text.encode(utf-8)).hexdigest() def run_pipeline(api_client, data_dir: Path, cache_dir: Path): cache_dir.mkdir(exist_okTrue) for image_path in sorted(data_dir.glob(*.jpg)): img_key md5_of_text(image_path.name) stage1_file cache_dir / f{img_key}_desc.json stage2_file cache_dir / f{img_key}_text.json stage3_file cache_dir / f{img_key}_triples.json if not stage1_file.exists(): vision_note describe_image(api_client, str(image_path)) stage1_file.write_text(json.dumps({vision_note: vision_note}, ensure_asciiFalse)) if not stage2_file.exists(): vision_note json.loads(stage1_file.read_text())[vision_note] specs load_specs_for_image(image_path) text generate_dynamic_text(api_client, inspection, vision_note, specs) stage2_file.write_text(json.dumps({text: text}, ensure_asciiFalse)) if not stage3_file.exists(): text json.loads(stage2_file.read_text())[text] triples extract_triples(api_client, text) stage3_file.write_text(json.dumps(triples, ensure_asciiFalse)) write_triples_to_neo4j(json.loads(stage3_file.read_text()), sourceimage_path.name)逻辑说明每个阶段先检查缓存文件是否存在存在就直接读不存在才调接口。这样批量处理几百张图片时中间接口抖一下、网络抖一下重新执行脚本只会从失败的那个文件继续。断点是整个 pipeline 设计里被低估的一环没有断点的脚本跑半小时挂在中间只能从头再来这种血泪教训一次就够。参数说明缓存文件的命名用的是图片名 MD5避免文件名带中文或特殊字符引发路径问题。load_specs_for_image 是业务侧函数根据图片名从参数表里找对应的结构化数据这里省略了实现。write_triples_to_neo4j 内部用上一章的 Cypher 模板统一处理导入逻辑。5.2 避坑记录一图谱里出现了大量重复节点现象跑完 pipeline 后查 count发现“空压机-01”节点出现十几次数据量虚高查询结果全乱。原因抽取阶段模型输出了“空压机-01”和“空压机 - 01”两种写法实体规范化没做。MERGE 判重基于 name 精确匹配空格差异直接绕过判重。解决入库前增加实体规范化函数去掉所有空白字符、全角转半角、统一大小写。我在规范化里还加了一个同义词映射表把“空压机1号”“空压机 1 号”这类写法统一归一到“空压机-01”。从那以后图谱节点数量稳定在预期范围不再虚增。5.3 避坑记录二批量调用 DeepSeek 时频繁超时现象单张图片调用正常批处理跑 200 张时每隔几分钟就抛一次超时异常中断整个 pipeline。原因循环里同步调用接口上一请求还没返回就发下一个服务端并发压力大接口响应时间被拉长本地网络环境也不是专线偶发丢包会被超时放大。解决限制并发数用信号量控制同时进行中的请求不超过 5 个超时时间从默认 60 秒调到 120 秒增加指数退避重试第一次失败等 2 秒重试第二次等 4 秒。改完后批处理稳定跑完超时率从百分之十几降到接近零。5.4 避坑记录三三元组解析失败导致入库条数大幅缩水现象某批次 500 条文本入库的三元组只有 200 多条大量候选关系被丢弃。原因抽取提示词里没限制“不要输出解释”模型在 JSON 数组前后加了“以下是抽取结果”这类话严格 json.loads 失败后正则兜底无法匹配这些额外内容整段丢弃。解决在系统消息里加一句“只输出 JSON 数组任何解释都会被视为错误”并把温度从 0.1 下调到 0同时把正则兜底的模式改成允许前后存在任意非 JSON 字符。处理后该批次三元组数量回升到 450 条以上。5.5 避坑记录四Neo4j 批量导入时事务回滚现象一次导入 3000 行时报内存不足整个事务回滚中间结果全部丢失。原因UNWIND 一次展开 3000 行MERGE 创建的节点和关系对象全在事务内存里内存占用超限触发回滚。解决把导入改成每 500 行提交一次事务用循环分批。Cypher 本身不支持事务控制但事务边界由驱动 API 控制Python 驱动里每 500 行执行一次 session.execute_write 就是一个干净事务。批量导入不是一个事务越大越快合理批次反而更稳。5.6 边界条件什么场景不建议用这套 pipeline这套链路不是万能的。如果业务只有纯文本、没有图片多模态需求直接调 DeepSeek 文本接口做抽取就行引入图像描述阶段只会增加 latency 和成本。实体关系高度不固定的场景比如开放域问答里的即兴关系也不适合知识图谱——关系类型不收敛图谱会退化成一个大号 JSON 存储查询收益为零。成本也是边界之一。每张图至少产生两次模型调用图像描述一次、文本生成一次抽取可能还要第三次数据量上百万时这个成本必须提前评估。PDF 里给的 pipeline 设计更适用于数据万级到十万级的中型业务再大的规模要考虑离线处理或消息队列而不是单机脚本硬跑。6. 质量验证与回捞上线前先跑一遍覆盖率检查6.1 覆盖率与孤立节点检查脚本图谱导入完成后先别急着接业务查询。我习惯先跑一遍质量检查看看数据像不像样。检查脚本就几条简单 Cypher 统计def verify_graph(session): total_nodes session.run(MATCH (n) RETURN count(n) AS c).single()[c] orphan_nodes session.run( MATCH (n) WHERE NOT (n)--() RETURN count(n) AS c ).single()[c] relation_types session.run( MATCH ()-[r]-() RETURN type(r) AS t, count(r) AS c ).data() print(f总节点数: {total_nodes}, 孤立节点: {orphan_nodes}, 占比: {orphan_nodes / total_nodes:.2%}) for item in relation_types: print(f关系 {item[t]}: {item[c]})逻辑说明孤立节点占比是最直观的体检指标。如果超过 10%大概率是抽取阶段的关系类型没匹配上白名单导致边没建起来。关系类型分布能看出数据是否偏科——比如“关联备件”只有几十条而“发生故障”有几万条说明备件数据来源缺失后续要补采集。覆盖率检查通过后我才会把查询模板接到 Dashboard 上。6.2 回捞机制低置信度关系重新生成检查中发现某个关系类型数量异常少不用整批重新跑 pipeline。我的做法是定位到具体来源把对应文本单独捞出来换一个更低温度、更严格约束的提示词重新抽取只更新差异部分。回捞是增量操作涉及的数据量通常只有几百条成本可控。从那以后我每次跑完 pipeline 都强制走一遍覆盖率检查孤立节点比例超过阈值就直接回捞不再抱着侥幸心理往下走。质量检查这一关设好后面接 RAG 也好、接智能问答也好返回结果才不会时灵时不灵。希望这份从多模态输入到图谱入库的完整实现能帮你少烧几个晚上调试 prompt。本文还有配套的精品资源点击获取