简介：这份DeepSeek金融机构数据中台与知识图谱构建方案共522页，深度聚焦金融行业非结构化数据自动抽取与实体关系对齐知识图谱构建，适合数据架构师、AI算法工程师及金融科技从业者参考。文档基于DeepSeek-R1展开，系统覆盖从多源异构数据采集、文本/语音/图像预处理，到实体类型体系、属性定义、关系梳理及标注体系建设，再到实体抽取模型选型与提示词工程设计的完整技术链路，内容包含研报/合同/公告清洗去噪、客服录音转写标准化、财报扫描件OCR识别与精度优化、分布式集群部署与调优等具体实现方案。包体为单个PDF文件，大小15.25MB，内置52个章节，支持目录跳转和阅读器书签大纲快速定位，便于按模块查阅。目前已有83人学习。相比零散资料，这套方案能帮助读者理清金融数据中台与知识图谱从0到1的落地方案，并沉淀一套可复用的构建方法论，特别适合正在规划或实施金融知识图谱项目的团队作为技术蓝本。
1. DeepSeek 金融知识图谱：先解决“同一家公司被写成八个名字”
做金融机构的数据中台项目时，最常遇到的需求就是把年报、尽调纪要、合同扫描件这类非结构化文本，变成可查询、可下钻、可预警的结构化关系。过去用正则加关键词，抽取结果又碎又乱，同一家公司在不同报告里有七八种写法；DeepSeek 这类大模型把抽取的召回率带上一个台阶，但金融场景真正卡住的往往不是抽取，而是抽取之后的对齐——怎么把“中信证券”“CITIC Securities”“中信”指到同一个节点。
这篇博文按一条完整链路讲：DeepSeek 抽三元组、实体关系对齐、挂进数据中台分层存储、Neo4j 构建知识图谱并回哺宽表。每一步都给出最小可运行的代码和参数，适合正在做金融知识图谱、智能风控、授信关联排查的工程师和数据产品经理。
抽取、对齐、入库三个环节的顺序可以调，但前置条件不能省：抽取决定数据下限，对齐决定图正确性，入库方式决定查询能不能复用。
2. 用 DeepSeek 抽非结构化数据：分块、JSON 抽取与 ODS 暂存
2.1 定分块规则：一句话断在两块里，模型就会编关系
非结构化数据的载体大多是 PDF，第一步是文本析出。常见做法是用 PyMuPDF 取页面文本，pdfplumber 处理表格；如果扫描件没有文本层，先走 OCR。析出之后必须分块，分块不是均匀切页，而是按语义块切：把“甲方：XXXX 有限公司”和“乙方：XXXX 银行”放在同一块里，把一份授信合同里的金额、期限、担保条款放同一块。分块太碎，DeepSeek 看不到完整上下文，容易把“该公司”指到前一块的主体上。
下面这个函数按段切块，块的上限用字符数控制。中文一个字大约占 1.2 到 1.5 个 token，块设 1200 字，上下文窗口只用了四分之一左右，剩下留给提示词和输出。
import fitz def extract_and_chunk(pdf_path: str, max_chars: int = 1200) -> list[dict]: doc = fitz.open(pdf_path) text = "\n".join(page.get_text("text") for page in doc) blocks = [b.strip() for b in text.split("\n\n") if b.strip()] chunks, buf = [], "" for block in blocks: # 表头和金额连续行被拆开时强制合并 if buf and _belongs_to_prev(block, buf): buf += "\n" + block continue if len(buf) + len(block) > max_chars: chunks.append({"chunk_id": f"c{len(chunks):04d}", "text": buf}) buf = block else: buf += "\n" + block if buf: chunks.append({"chunk_id": f"c{len(chunks):04d}", "text": buf}) return chunks def _belongs_to_prev(block: str, prev: str) -> bool: # 当前块以“金额”“合计”开头，且前一块以“单位：”结尾时合并 return block[:2] in ("金额", "合计") and prev.rstrip().endswith("单位：")max_chars 按字符切，不按页切，避免一张表被拆到两个块。_belongs_to_prev是很弱的启发式，生产环境建议把“表头在页尾、表格在下一页”的情况统一按跨页表格合并处理，比让模型猜更可靠。分块之后要给每个块一个稳定 ID，后续 evidence 定位靠它。
2.2 最小可跑的 DeepSeek API 抽取代码
DeepSeek 的接口兼容 OpenAI 协议，用 openai SDK 就能调。关键点有两个：一是把输出限定成 JSON，二是把关系枚举写死在提示词里。金融实体的关系如果不枚举，模型会自由发挥出“参与”“涉及”这类语义含糊的边，图谱里出现几十种关系类型，后面查询和预警都很难做。
from openai import OpenAI import json, os client = OpenAI( api_key=os.environ["DEEPSEEK_API_KEY"], base_url="https://api.deepseek.com", # 本地部署时换成内网服务地址 ) PROMPT_TMPL = """你是金融知识图谱抽取器。从文本中抽取公司、自然人、金融产品三类实体， 以及以下五种关系之一：持有股权、授信、担保、实际控制人、母公司。 只输出 JSON，不要输出其他文字，格式： {"triples":[{"head":"","relation":"","tail":"","evidence":"原文片段"}]} 没有把握的实体不要输出，宁可漏抽不要编造。 文本： {chunk}""" def extract_triples(chunk: str) -> list[dict]: resp = client.chat.completions.create( model="deepseek-chat", messages=[{"role": "user", "content": PROMPT_TMPL.format(chunk=chunk)}], temperature=0, max_tokens=2000, response_format={"type": "json_object"}, ) return json.loads(resp.choices[0].message.content)["triples"]调用参数按抽取任务来定，和对话场景完全不同。下面这组设置按结果一致性和可解析性优先。
| 参数 | 建议值 | 说明 |
|---|---|---|
| temperature | 0 | 抽取是判别任务，温度越高越容易编造不存在的担保关系 |
| max_tokens | 1500–3000 | 和分块大小正相关，截断会导致 JSON 解析失败 |
| response_format | {"type":"json_object"} | DeepSeek 的 JSON 模式，提示词里必须出现“JSON”字样 |
| timeout / retries | 60 秒 / 3 次 | 长块推理耗时高，指数退避重试 |
2.3 抽取结果先落 ODS 暂存表，别直接进图
抽出来的三元组不要直接写进 Neo4j。金融场景要给每条关系留证据和批次，否则后续对齐错了没法回溯。常见做法是先落到 ODS 层暂存表，按天分区，重跑时按 batch_id 清数。
CREATE TABLE ods_deepseek_triple_stage ( batch_id STRING COMMENT '抽取批次，格式: yyyymmdd_hhmm', chunk_id STRING COMMENT '来源分块ID，定位原文用', head STRING COMMENT '头实体原文', relation STRING COMMENT '关系，限枚举值', tail STRING COMMENT '尾实体原文', evidence STRING COMMENT '原文证据句，对齐和审计必留', model_name STRING COMMENT 'deepseek-chat 或本地部署的模型名', created_at TIMESTAMP ) PARTITIONED BY (dt STRING);evidence 字段是对齐阶段最重要的输入，不能省；head 和 tail 存的是原文写法，不是最终实体 ID，规范化在下一层做。model_name 记录是哪一版模型抽的，模型升级后可以用同批文本做回归对比。
提示：数据不能出域的机构，常见做法是本地部署 DeepSeek，用 vLLM 或 Ollama 起一个 OpenAI 兼容服务。上面的代码只需要改 base_url 和 model 名，表结构和提示词原样保留。
3. 实体关系对齐的三层方案：规范化、别名表与相似度窗口
3.1 对齐错误会顺着图谱传染
抽取错误是局部问题，一条关系错了，影响的是单个边；对齐错误是全局问题，两个实体被错误合并，所有经过这两个节点的传导路径都会串起来。金融机构做风险传导分析时，把两家名称相近但主体不同的公司合并，会把不存在的担保链画出来。所以图谱质量的第一指标不是抽取精度，而是对齐精度。
对齐要区分实体类型，不能一套规则打天下。公司实体靠名称规范化加社会信用代码；自然人靠证件号脱敏哈希；金融产品靠产品代码和登记编码。名称相似度只是兜底，永远排在证件和代码映射之后。下面按公司实体讲一套分层方案，自然人流程类似，只是指纹函数换成证件号哈希加姓名窗口。
3.2 三层对齐策略：L1 规范化、L2 别名表、L3 上下文消歧
| 层级 | 手段 | 示例 |
|---|---|---|
| L1 规范化 | 去空格、统一全半角括号、去公司类型后缀 | “中信证券股份有限公司” → “中信证券” |
| L2 别名表 | 显式映射中英文、简称、曾用名 | CITIC Securities / 中信 → 主体 ID |
| L3 上下文消歧 | 用证据里的行业、地域、时间窗判断 | 报告里的“中信”前面出现过“银行”则归中信银行 |
L1 是提效的，能减少七成候选对；L2 是兜底的，解决目公司代码和曾用名这类规范化处理不了的问题；L3 是收口的，处理规范化后仍然重名、但实际是不同主体的案例。“中信”到底是集团、证券还是银行，单看字符串判断不了，必须看证据上下文。L3 消歧可以用规则，也可以再让 DeepSeek 做一次指代消解，但注意消解输出仍要限定在关系枚举内。
3.3 用归一化指纹加相似度窗口跑一个可复现的对齐
对齐必须可复现，不建议直接在界面上手工合并。常见做法是把候选对按批次抽出来，用归一化指纹分桶，在桶内做相似度打分，输出候选清单。指纹相同的直接标记为同一主体；指纹不同但相似度高于阈值的进入人工审核；低于阈值的靠别名表和 L3 兜底。
import re, difflib from collections import defaultdict SUFFIX = re.compile(r"(股份有限公司|有限责任公司|有限公司|集团|\(.*?\)|（.*?）)") def norm_company(name: str) -> str: name = name.strip().upper() name = re.sub(r"[\s'\"（）\(\)\-·_]", "", name) return SUFFIX.sub("", name) def candidate_pairs(rows: list[dict], review_lo: float = 0.90, auto_hi: float = 0.96) -> list[dict]: buckets = defaultdict(list) for r in rows: key = norm_company(r["head"]) if key: buckets[key].append(r) out = [] keys = list(buckets) for i in range(len(keys)): for j in range(i + 1, len(keys)): left, right = keys[i], keys[j] score = difflib.SequenceMatcher(None, left, right).ratio() if score >= review_lo: out.append({ "left": buckets[left][0], "right": buckets[right][0], "score": round(score, 4), "auto_merge": score >= auto_hi, }) return out参数有三个注意点。review_lo 定 0.90，低于这个分数的桶大概率不是同一主体，交给别名表处理；auto_hi 定 0.96，只有去掉公司类型后缀后仍然高度一致的才自动合并，例如“中信证券”和“中信证券股份”。0.90 到 0.96 之间全部落人工审核，宁可多审，不要在生产流程里直接合并。difflib 的 ratio 对字符顺序敏感，“XX 银行”和“银行 XX”这种写法会漏，正规化阶段要把明显的倒序写法归一下。
3.4 对齐决策要留证据：matched_by 与 operator
合并是一个决策，决策必须留痕。对齐结果表加四个字段：matched_by 记录走的是哪一层（norm / alias / fuzzy / manual），evidence 记录两边各自的关键证据，operator 记录审核人，version 记录对齐规则版本。这样做的用途是：图谱重放时按 version 取最新决策，历史决策不删除只标记过期，审计时能讲清楚“为什么把这两个名字合并了”。
4. 数据中台落地知识图谱：冷热分层、Neo4j 导入与宽表回哺
4.1 图谱层挂在中台哪一层：ODS、DWD、图库与 ADS
数据中台里建知识图谱，不是把 Neo4j 当成新的数据湖。常见架构是：ODS 存 DeepSeek 抽出的三元组暂存；DWD 层放对齐后的主体维表和关系事实表；Neo4j 从 DWD 读取加工好的实体与关系，承担在线多跳查询；ADS 层放从图谱导出的宽表和指标。图库在这套结构里是中间计算引擎，不是源系统。
这个分层的好处是图库可以随时重建。DWD 的关系事实表是唯一事实来源，Neo4j 只是它的投影；模型升级导致抽取结果变化时，重跑 ODS 和 DWD，再全量重建图库，不会出现图库和数仓对不上的问题。血缘也清晰：从证据文本到实体 ID 再到图里的关系属性，每一跳都有表可查。
4.2 冷热数据分治：活跃实体进在线库，快照进归档表
图谱数据有两个特点：一是随时间增长，快照和变更记录无限膨胀；二是查询热度极不均匀，绝大多数下钻都集中在近期有事件的实体上。数据中台的冷热数据不能混在一张表里，图库也一样。常见做法是两条路径分开：
| 数据 | 存储位置 | 使用场景 |
|---|---|---|
| 活跃实体、近 12 个月关系 | Neo4j 在线库 | 页面下钻、实时预警 |
| 季度全量快照、变更日志、对账结果 | 归档表 / 对象存储 | 回溯、离线分析、特征训练 |
归档表的设计要点是只追加不更新。每个季度导一份全量快照，写一条快照元数据；审计要求“某年某月图上是什么样”时，从对应快照恢复，而不是指望在线库保留历史版本。
CREATE TABLE dwd_kg_snapshot_archive ( snapshot_id STRING COMMENT '快照批次', entity_id STRING, entity_name STRING, entity_type STRING, relation_to STRING COMMENT '边终点实体ID', relation_type STRING ) PARTITIONED BY (quarter STRING);4.3 Neo4j 构建知识图谱：先建约束，再周期提交导入
从 DWD 导出 CSV 后导入 Neo4j，顺序是先建约束和索引，再导数据。约束保证 corp_id 唯一，防止同一主体被插成多个节点；索引保证按名称检索不扫全库。导边时用 MERGE 而不是 CREATE，同一批数据重复导入不会生成重复边。
CREATE CONSTRAINT company_id IF NOT EXISTS FOR (c:Company) REQUIRE c.corp_id IS UNIQUE; CREATE INDEX company_name_idx IF NOT EXISTS FOR (c:Company) ON (c.name); :auto USING PERIODIC COMMIT 500 LOAD CSV WITH HEADERS FROM 'file:///dwd_kg_rel.csv' AS row MATCH (h:Company {corp_id: row.head_id}) MATCH (t:Company {corp_id: row.tail_id}) MERGE (h)-[r:REL {type: row.relation}]->(t) SET r.evidence = row.evidence, r.batch_id = row.batch_id;两个 MATCH 都要求端点 corp_id 在库里存在，CSV 里残留的孤儿行会被跳过，导入日志里会记 warning，处理完再决定补主体还是删关系。PERIODIC COMMIT 把事务拆小，单批次 500 行，千万级边也能跑完。关系边上带 evidence 和 batch_id，在图里点一条边就能看到它来自哪个批次、哪句原文，风控场景审计时直接可用。
如果一条边的端点可能是公司也可能是自然人，不能只写一个 MATCH。要分两段 MATCH（先 Company 后 Person），或者拆成两个导入脚本再合并；端点类型有歧义时，宁可拆脚本也不要建没有类型约束的通用节点。
4.4 查询结果反哺宽表：在线多跳与批量展开分开做
图谱搭好之后还要把查询结果回写到 ADS 宽表，否则报表、模型和预警规则都实时查图，Neo4j 扛不住。在线下钻查询用图库的多跳能力：
MATCH p = (:Company {corp_id: $corpId})-[:REL*1..4]->(x:Company) RETURN x.corp_id, x.name, reduce(rs = '', r IN relationships(p) | rs + r.type + '>') AS path全市场传导扫描这类批处理任务，不在在线库跑，而是把 DWD 关系事实表展开成宽表，按“起点实体、终点实体、中间路径数、关联担保金额”这类指标落 ADS。在线查询和批量展开分开，是知识图谱在数据中台里能长期跑下去的关键。
5. 对齐抽检脚本：把 0.90–0.96 的候选对导出成可审 JSON
5.1 抽检脚本别在界面上点，导成 JSONL 交给复核
对齐候选如果直接在管理界面上人工点，容易点错还没有痕迹。我习惯按批次导出 JSONL，每条记录带左右两边的实体 ID、名称、证据原文和相似度分数，复核人在文件里填 action 字段，回写后归档。这个文件本身就是审计证据。
import json, sqlite3 def dump_review_queue(db_path: str, out_path: str, limit: int = 200): conn = sqlite3.connect(db_path) rows = conn.execute(""" SELECT left_id, right_id, score, l_name, r_name, l_evidence, r_evidence FROM alignment_candidate WHERE score BETWEEN 0.90 AND 0.96 AND review_status = 'pending' ORDER BY score DESC LIMIT ? """, (limit,)).fetchall() with open(out_path, "w", encoding="utf-8") as f: for left_id, right_id, score, ln, rn, lev, rev in rows: f.write(json.dumps({ "left": {"id": left_id, "name": ln, "evidence": lev}, "right": {"id": right_id, "name": rn, "evidence": rev}, "score": score, "action": None # 复核人填 "merge" 或 "reject" }, ensure_ascii=False) + "\n") conn.close()复核完成后把 action 回写 alignment_decision 表，带 version 字段；下一次图谱重放只认最新 version，旧决策留档不覆盖。即便某一次批量合并判断错了，也能定位到是哪个批次、谁审的、依据是哪条 evidence，改完重放一个版本即可。
5.2 反向校验：用向量近邻回填别名表
人工复核永远覆盖不全，还要一条自动反向校验。每两周跑一次：把已经合并过的实体对取出双方名称和证据文本，用 Embedding 模型编码后做近邻检索。如果出现“向量距离很近且没有被任何规则命中”的对，说明别名表漏了写法，把这对写法回填到 L2 别名表，下一次对齐就能直接命中。
跑完这轮抽检，把对齐候选表的 pending 状态清掉，再把新增的别名映射提交到规则仓库，和提示词文件放一起管理。下次再跑对齐时，抽检队列、别名映射和提示词三个文件打同一个版本号再提交，图谱质量就跟着数据中台版本一起迭代了。
本文还有配套的精品资源，点击获取
