基于Python的豆瓣书籍与电影知识图谱问答系统构建实战

简介这是一套基于Python的豆瓣书籍与电影类别知识图谱问答系统完整项目包面向计算机、数学、电子信息等专业的学生与开发者适合用作课程设计、期末大作业或毕业设计的学习参考。资源包含全部源码、数据库、数据集与示例图片下载后可直接运行调试。压缩包共1394个文件约58.28MB以html页面、java源码、rq查询脚本、js与css前端资源为主同时包含ttl、n3、rdf、owl等本体与三元组数据文件以及py脚本、sql建库脚本、bat启动脚本和jar依赖包覆盖知识图谱构建、存储与问答交互的完整链路。目前已有287人学习下载。项目围绕豆瓣书籍与电影两类实体提供从数据抽取、图谱存储到自然语言问答的实现思路读者可据此理解本体设计、SPARQL查询与前后端联调方式并在此基础上扩展新的实体类别或问答功能。1. 从豆瓣书籍与电影数据出发知识图谱问答系统到底解决什么问题豆瓣上的书籍和电影数据天然就是一张巨大的关系网。一本书有作者、出版社、出版年份、ISBN、评分、标签一部电影有导演、编剧、主演、上映日期、类型、评分、短评。当你想查“刘慈欣写的、评分超过9分、被改编成电影的作品有哪些”时传统数据库需要多表联查写一长串 SQL而知识图谱可以把这些实体和关系直接存成图结构用一次查询就能拿到答案。基于 Python 的豆瓣书籍与电影类别知识图谱问答系统核心就是把这层关系抽出来、存进去、查出来再包一层自然语言问答接口。它适合有 Python 基础、想入门知识图谱构建与问答系统落地的开发者也适合需要快速搭建垂直领域问答原型的团队。整套方案通常包含数据采集与清洗、本体设计、图谱构建、数据库存储、问答解析与前端展示几个环节源码、数据库文件、示例数据和图片就是围绕这些环节打包的。2. 知识图谱构建从豆瓣原始数据到三元组2.1 本体设计先定实体和关系再谈存储知识图谱不是先把数据扔进数据库再想怎么连而是先确定本体。豆瓣书籍和电影场景下我一般会抽这几类实体书籍、电影、人物、出版社、类型标签。关系则包括书籍-作者-人物、书籍-出版社-出版社、书籍-标签-类型、电影-导演-人物、电影-主演-人物、电影-类型-类型、电影-改编自-书籍。这个本体不用太复杂够覆盖常见问答就行。比如“某导演还导过哪些电影”“某本书有没有被改编成电影”“某标签下评分最高的书”这三类问题分别对应导演关系、改编关系、标签加评分属性。本体确定后后续的 CSV 或 JSON 数据字段就要按这个结构对齐避免后期返工。2.2 数据清洗把半结构化文本变成规整三元组豆瓣页面抓下来的数据往往带有多余空格、中文括号、多个作者用斜杠分隔等情况。清洗的目标是让每条记录都能稳定拆成主语、谓语、宾语。下面这段 Python 代码演示了从原始字典列表生成三元组的过程输入假设是已经解析好的书籍和电影记录。import re def clean_text(text): 去掉首尾空白统一括号和分隔符 if not text: return text text.strip() text text.replace(, ().replace(, )) text re.sub(r\s, , text) return text def split_multi(value, sep/): 把 刘慈欣 / 李淼 拆成列表并过滤空串 if not value: return [] parts [clean_text(p) for p in value.split(sep)] return [p for p in parts if p] def book_to_triples(book): 单本书记录转三元组列表 triples [] title clean_text(book.get(title)) if not title: return triples # 书籍本身属性 triples.append((title, 类型, 书籍)) if book.get(publisher): triples.append((title, 出版社, clean_text(book[publisher]))) if book.get(pub_year): triples.append((title, 出版年份, str(book[pub_year]))) if book.get(rating): triples.append((title, 评分, str(book[rating]))) # 作者关系 for author in split_multi(book.get(author, )): triples.append((title, 作者, author)) # 标签关系 for tag in split_multi(book.get(tags, )): triples.append((title, 标签, tag)) return triples def movie_to_triples(movie): 单部电影记录转三元组列表 triples [] title clean_text(movie.get(title)) if not title: return triples triples.append((title, 类型, 电影)) if movie.get(release_date): triples.append((title, 上映日期, clean_text(movie[release_date]))) if movie.get(rating): triples.append((title, 评分, str(movie[rating]))) for director in split_multi(movie.get(director, )): triples.append((title, 导演, director)) for actor in split_multi(movie.get(actors, )): triples.append((title, 主演, actor)) for genre in split_multi(movie.get(genres, )): triples.append((title, 类型标签, genre)) # 改编自某本书字段名可能是 adapted_from if movie.get(adapted_from): triples.append((title, 改编自, clean_text(movie[adapted_from]))) return triples这段代码的关键点在于clean_text统一了全角半角括号和多余空白避免同一个实体因为空格差异被当成两个节点split_multi处理了豆瓣常见的斜杠分隔多值字段book_to_triples和movie_to_triples分别把书籍和电影记录映射成统一的三元组格式。参数上sep默认是斜杠如果实际数据用逗号或顿号需要对应调整。评分字段建议统一转成字符串再入库避免浮点精度导致查询时匹配不上。2.3 三元组去重与冲突处理同一本书可能出现在多个数据源里作者字段一个写“刘慈欣”另一个写“刘慈欣 / 李淼”直接合并会产生重复边。常见做法是先用(主语, 谓语, 宾语)做一次集合去重再对实体名做归一化。比如把“刘慈欣”和“刘慈欣 ”统一把“科幻”和“科幻小说”根据业务需要决定是否合并。如果同一对主语谓语出现多个不同宾语比如一本书有两个出版社那就保留多条边不要强行覆盖。这一步做完三元组数量通常会比原始记录条数多出三到五倍属于正常现象。3. 用 Neo4j 还是 SQLite存储选型与入库实操3.1 图数据库和关系型数据库的取舍知识图谱问答系统不一定非得上 Neo4j。如果数据量在几万条三元组以内SQLite 完全够用而且部署简单一个文件就能带走。但一旦涉及多跳查询比如“找出所有被改编成电影、且电影评分高于书籍评分的书籍”SQL 写起来会非常痛苦而 Cypher 几行就能搞定。我的建议是如果只是做课程设计或小规模演示SQLite 加一张三元组表足够如果问答里经常出现两跳以上的关系查询直接上 Neo4j。下面分别给出两种入库方式的核心代码。3.2 SQLite 三元组表入库import sqlite3 def init_sqlite(db_pathdouban_kg.db): conn sqlite3.connect(db_path) cur conn.cursor() cur.execute( CREATE TABLE IF NOT EXISTS triples ( id INTEGER PRIMARY KEY AUTOINCREMENT, subject TEXT NOT NULL, predicate TEXT NOT NULL, object TEXT NOT NULL ) ) cur.execute(CREATE INDEX IF NOT EXISTS idx_subject ON triples(subject)) cur.execute(CREATE INDEX IF NOT EXISTS idx_predicate ON triples(predicate)) cur.execute(CREATE INDEX IF NOT EXISTS idx_object ON triples(object)) conn.commit() return conn def insert_triples(conn, triples): cur conn.cursor() cur.executemany( INSERT INTO triples (subject, predicate, object) VALUES (?, ?, ?), triples ) conn.commit()init_sqlite建了一张三元组表并对主语、谓语、宾语分别建索引这样按实体名查关系时不会全表扫描。insert_triples用executemany批量插入比逐条插入快一个数量级。参数上db_path可以指向打包好的数据库文件如果已有数据直接连接即可不需要重复建表。3.3 Neo4j 批量导入与 Cypher 查询如果选 Neo4j可以用 Python 驱动逐条写入也可以先把三元组导成 CSV再用LOAD CSV批量导入。下面是用官方驱动写入的示例。from neo4j import GraphDatabase class Neo4jWriter: def __init__(self, uri, user, password): self.driver GraphDatabase.driver(uri, auth(user, password)) def close(self): self.driver.close() def insert_triple(self, subject, predicate, obj): # 用 MERGE 避免重复创建节点 query MERGE (a:Entity {name: $subject}) MERGE (b:Entity {name: $object}) MERGE (a)-[r:REL {type: $predicate}]-(b) with self.driver.session() as session: session.run(query, subjectsubject, predicatepredicate, objectobj) def batch_insert(self, triples): with self.driver.session() as session: for s, p, o in triples: session.run( MERGE (a:Entity {name: $s}) MERGE (b:Entity {name: $o}) MERGE (a)-[:REL {type: $p}]-(b), ss, pp, oo )MERGE是这里的关键它保证同一个实体名不会重复建节点同一条关系也不会重复建边。batch_insert逐条执行适合几千到几万条数据如果上十万条建议改成LOAD CSV加apoc.periodic.iterate。参数上uri一般是bolt://localhost:7687user和password按实际配置填。写入完成后可以用一条 Cypher 验证“刘慈欣”相关的所有关系。MATCH (a:Entity {name: 刘慈欣})-[r:REL]-(b) RETURN a.name, r.type, b.name LIMIT 203.4 问答查询的两种实现路径问答系统的查询层有两种常见做法。一种是模板匹配预先定义“谁导演了某电影”“某作者写了哪些书”等句式解析出实体和意图后拼 Cypher 或 SQL。另一种是语义解析用模型把自然语言转成查询语句。对于豆瓣书籍电影这种领域窄、关系固定的场景模板匹配加同义词扩展就能覆盖大部分问题开发成本低可控性强。比如用户问“刘慈欣有哪些书”先识别出实体“刘慈欣”和意图“作者作品”然后执行MATCH (a:Entity {name:刘慈欣})-[:REL {type:作者}]-(b) RETURN b.name。如果问“评分最高的科幻书”则识别出标签“科幻”和排序意图执行带ORDER BY的查询。4. 问答系统落地意图识别、查询生成与前端展示4.1 意图分类与实体抽取的最小实现意图分类不需要一上来就上 BERT。对于豆瓣书籍电影问答问题类型通常集中在十几种以内用关键词加正则就能做到八成准确率。下面是一个轻量级的意图识别函数。import re INTENT_PATTERNS [ (author_works, r(.*)写的书|(.*)的作品|(.*)有哪些书), (director_movies, r(.*)导演的电影|(.*)导过哪些), (movie_actors, r(.*)的主演|(.*)是谁演的), (book_movies, r(.*)改编的电影|(.*)拍成了什么), (top_rated, r评分最高|评分前几|最受欢迎), ] def detect_intent(question): for intent, pattern in INTENT_PATTERNS: m re.search(pattern, question) if m: # 取第一个非空分组作为实体 entity next((g for g in m.groups() if g), ) return intent, entity.strip() return unknown, INTENT_PATTERNS里每条规则对应一种问法detect_intent返回意图标签和抽取到的实体。参数上正则里的(.*)是贪婪匹配如果实体名本身包含“的”字可能会多截实际使用时可以改成非贪婪或加边界。这个函数适合快速跑通流程后续要提升准确率可以换成基于标注数据的文本分类模型。4.2 查询模板与结果组装拿到意图和实体后下一步是生成查询。以 SQLite 为例作者作品查询可以这样写。def build_sql(intent, entity): if intent author_works: return ( SELECT object FROM triples WHERE predicate作者 AND subject IN (SELECT subject FROM triples WHERE predicate类型 AND object书籍) AND object?, (entity,) ) if intent director_movies: return ( SELECT subject FROM triples WHERE predicate导演 AND object?, (entity,) ) if intent book_movies: return ( SELECT subject FROM triples WHERE predicate改编自 AND object?, (entity,) ) return None, None这里用子查询先筛出书籍类实体再匹配作者关系避免把电影导演也当成作者返回。build_sql返回参数化查询防止拼接字符串带来的注入问题。执行后把结果列表拼成自然语言回复即可。如果用的是 Neo4j把 SQL 换成对应的 Cypher 模板逻辑一致。4.3 前端展示最小可用页面前端不需要太复杂一个输入框加一个结果区域就能演示。用 Flask 起一个接口前端用原生 fetch 调用。from flask import Flask, request, jsonify app Flask(__name__) app.route(/ask, methods[POST]) def ask(): question request.json.get(question, ) intent, entity detect_intent(question) if intent unknown: return jsonify({answer: 暂时无法理解这个问题}) sql, params build_sql(intent, entity) if not sql: return jsonify({answer: 暂不支持该查询}) conn sqlite3.connect(douban_kg.db) cur conn.cursor() cur.execute(sql, params) rows [r[0] for r in cur.fetchall()] conn.close() if not rows: return jsonify({answer: f没有找到与「{entity}」相关的结果}) return jsonify({answer: 、.join(rows[:10])})这个接口把意图识别、查询生成、数据库查询串起来返回前十条结果。参数上question从请求体 JSON 里取前端对应发送{question: 刘慈欣写的书}。如果结果太多可以加分页或限制条数避免前端渲染卡顿。5. 避坑与排查豆瓣知识图谱问答系统最常见的五个翻车点5.1 实体名不统一导致查不到结果现象明明数据库里有“刘慈欣”用户输入“刘慈欣”却返回空。原因入库时作者字段带了多余空格或全角括号比如“刘慈欣三体作者”而查询时用的是干净字符串。解决入库前统一做clean_text查询时也对实体做同样清洗或者建一张别名表把常见变体映射到标准名。5.2 多值字段拆分后产生重复边现象同一本书的标签“科幻”出现两次图谱里就有两条相同边。原因原始数据里标签字段可能重复或者清洗时没有去重。解决在生成三元组后用set(triples)去重再入库。如果用的是 Neo4j 的MERGE重复边不会产生但节点属性可能被覆盖需要留意。5.3 意图识别把电影问题当成书籍问题现象用户问“刘慈欣导演的电影”系统返回了刘慈欣写的书。原因意图规则里“刘慈欣”匹配到了author_works没有区分“写的”和“导演的”。解决调整正则优先级把更具体的模式放在前面或者在实体抽取后加一步类型判断看实体在库里主要关联的是书籍还是电影。5.4 SQLite 查询大数据量时变慢现象三元组超过十万条后按实体名查询要好几秒。原因虽然建了索引但LIKE或子查询嵌套太深。解决尽量用等值查询代替LIKE把子查询拆成两步先在 Python 里查出中间结果再查第二步。如果还是慢考虑换 Neo4j 或加缓存。5.5 Neo4j 批量写入时内存溢出现象一次性MERGE几万条边驱动报内存错误。原因每条session.run都开一个事务累积太多。解决改用LOAD CSV加apoc.periodic.iterate每批一千条提交一次或者用UNWIND把列表展开一次事务处理一批。6. 进阶技巧用规则加缓存把问答响应压到毫秒级模板匹配的问答系统瓶颈往往不在查询本身而在每次都要走一遍意图识别和数据库连接。我的习惯是加一层内存缓存把高频问题的结果直接存下来。比如“刘慈欣写的书”“评分最高的科幻书”这类问题第一次查完就把(question, answer)存进字典下次同样问题直接返回。缓存键可以用意图加实体拼成比如author_works:刘慈欣。这样命中缓存时响应时间从几十毫秒降到微秒级。另一个技巧是把同义词扩展做成配置表。用户可能问“刘慈欣的作品”“刘慈欣写的书”“刘慈欣有哪些书”三种问法对应同一个意图。与其写三条正则不如先做一次同义词替换把“作品”“写的书”统一成“书”再走一条规则。配置表可以放在 JSON 文件里方便后续增删。验证方法上我一般会准备二十到三十个典型问题覆盖作者、导演、改编、评分排序几类每次改完代码跑一遍看命中率和返回结果是否正确。如果某类问题频繁失败再针对性调整规则。这套系统不值得追求大而全能把豆瓣书籍电影场景下最常见的十几类问题答准就已经有实用价值了。希望帮到你。本文还有配套的精品资源点击获取