简介这是一套基于Python构建的豆瓣书籍与电影类别知识图谱问答系统完整项目包面向计算机、数学、电子信息等专业的学生与开发者可作为课程设计、期末大作业或毕业设计的参考资料帮助理解知识图谱从数据存储到智能问答的完整链路。压缩包共1394个文件约58.28MB涵盖545个html页面、316个java源码、146个rq查询脚本、46个js脚本、26个py文件及34个ttl本体文件并配有sql数据库、rdf/owl/n3等语义数据、bat启动脚本与png示例图片完整呈现前端展示、后端逻辑与图谱查询的协同结构。项目已吸引286人学习下载读者可从中获取可运行的源码、数据库与示例数据参考SPARQL查询与Jena/Fuseki服务配置快速搭建书籍与电影领域的问答演示并在此基础上进行功能扩展与二次调试。1. 从豆瓣书籍和电影数据出发一套能跑起来的 Python 知识图谱问答系统长什么样很多人第一次接触知识图谱是从 Neo4j 的官方示例或者某个图数据库的演示视频开始的看完觉得概念不难但真到自己动手面对一堆 CSV 和 JSON 就不知道从哪下手。这套基于 Python 的豆瓣书籍加电影类别知识图谱问答系统解决的正是这个断层它把豆瓣上抓下来的书籍、电影、导演、演员、作者、出版社、标签这些实体用 Python 清洗成结构化数据灌进图数据库再在上面搭一层自然语言问答接口。你输入“刘慈欣写了哪些书”或者“诺兰导演的电影评分最高的是哪部”系统解析意图、生成查询、返回答案。适合谁适合正在做数据库课程设计、想找一个完整知识图谱项目练手的学生也适合已经会 Python 基础语法、想搞明白知识图谱从数据到问答全链路的开发者。整套东西的核心不是算法多深而是数据建模和查询映射这两步能不能走通。2. 数据层怎么搭豆瓣书籍与电影实体的抽取、清洗和入库2.1 实体和关系的设计决定了后面问答能问什么知识图谱的骨架是本体设计也就是你打算把哪些东西当实体、哪些东西当关系。豆瓣数据天然适合拆成两类域书籍域和电影域。书籍域里核心实体是 Book、Author、Publisher、Tag电影域里核心实体是 Movie、Director、Actor、Genre。跨域实体是 Person因为一个人可能既是某本书的作者又是某部电影的导演或演员。关系设计上书籍侧常见的有(Author)-[:WROTE]-(Book)、(Publisher)-[:PUBLISHED]-(Book)、(Book)-[:HAS_TAG]-(Tag)电影侧有(Director)-[:DIRECTED]-(Movie)、(Actor)-[:ACTED_IN]-(Movie)、(Movie)-[:BELONGS_TO]-(Genre)。这里有个容易翻车的地方豆瓣的“作者”字段经常混着译者、编者如果你不区分后面问“某人的作品”就会把译者和原作者混在一起。我一般会在清洗阶段加一个 role 字段把作者、译者、编者分开存关系类型也拆成WROTE、TRANSLATED、EDITED。节点属性方面Book 至少要有 title、isbn、pub_year、rating、rating_countMovie 要有 title、release_year、rating、rating_count、duration。这些属性不只是展示用问答系统里“评分最高的电影”“2010 年以后出版的书”这类条件查询全靠它们。2.2 用 Python 把原始数据洗成 Neo4j 能吃的格式假设你手里已经有一份豆瓣书籍和电影的原始数据格式可能是 CSV 或者 JSON。下面这段代码做的是把原始记录拆成节点文件和关系文件输出成 Neo4j 的 LOAD CSV 能直接读的格式。import csv import json from collections import defaultdict # 读取原始豆瓣数据假设是 JSON 行格式 def load_raw(path): records [] with open(path, r, encodingutf-8) as f: for line in f: line line.strip() if line: records.append(json.loads(line)) return records # 拆分实体和关系 def build_graph_data(records): nodes defaultdict(dict) # {label: {id: {props}}} rels [] # [(start_label, start_id, rel_type, end_label, end_id)] for rec in records: if rec.get(type) book: bid fbook_{rec[id]} nodes[Book][bid] { title: rec.get(title, ), isbn: rec.get(isbn, ), pub_year: rec.get(pub_year, ), rating: rec.get(rating, 0), rating_count: rec.get(rating_count, 0) } # 作者关系区分角色 for a in rec.get(authors, []): aid fperson_{a[id]} nodes[Person][aid] {name: a[name]} rel_type WROTE if a.get(role) author else TRANSLATED rels.append((Person, aid, rel_type, Book, bid)) # 出版社 if rec.get(publisher): pid fpub_{rec[publisher]} nodes[Publisher][pid] {name: rec[publisher]} rels.append((Publisher, pid, PUBLISHED, Book, bid)) # 标签 for tag in rec.get(tags, []): tid ftag_{tag} nodes[Tag][tid] {name: tag} rels.append((Book, bid, HAS_TAG, Tag, tid)) elif rec.get(type) movie: mid fmovie_{rec[id]} nodes[Movie][mid] { title: rec.get(title, ), release_year: rec.get(year, ), rating: rec.get(rating, 0), rating_count: rec.get(rating_count, 0), duration: rec.get(duration, 0) } for d in rec.get(directors, []): did fperson_{d[id]} nodes[Person][did] {name: d[name]} rels.append((Person, did, DIRECTED, Movie, mid)) for a in rec.get(actors, []): aid fperson_{a[id]} nodes[Person][aid] {name: a[name]} rels.append((Person, aid, ACTED_IN, Movie, mid)) for g in rec.get(genres, []): gid fgenre_{g} nodes[Genre][gid] {name: g} rels.append((Movie, mid, BELONGS_TO, Genre, gid)) return nodes, rels # 输出成 CSV def export_csv(nodes, rels, out_dir./graph_csv): import os os.makedirs(out_dir, exist_okTrue) for label, items in nodes.items(): if not items: continue # 收集所有出现过的属性键 keys set() for props in items.values(): keys.update(props.keys()) keys sorted(keys) with open(f{out_dir}/nodes_{label}.csv, w, newline, encodingutf-8) as f: writer csv.writer(f) writer.writerow([id] keys) for nid, props in items.items(): writer.writerow([nid] [props.get(k, ) for k in keys]) with open(f{out_dir}/relationships.csv, w, newline, encodingutf-8) as f: writer csv.writer(f) writer.writerow([start_label, start_id, rel_type, end_label, end_id]) for r in rels: writer.writerow(r) if __name__ __main__: raw load_raw(douban_raw.jsonl) nodes, rels build_graph_data(raw) export_csv(nodes, rels) print(f节点标签数: {len(nodes)}, 关系数: {len(rels)})这段代码的逻辑分三步先按记录类型判断是书还是电影再分别抽取实体和关系最后统一导出。build_graph_data里用defaultdict(dict)存节点好处是同一个实体被多次引用时自动去重比如同一个作者写了多本书Person 节点只会保留一份。关系用列表存每条关系记录起点标签、起点 ID、关系类型、终点标签、终点 ID这样导入 Neo4j 时不用再猜。参数上要注意几个点role字段的取值决定了关系类型如果你的原始数据里没有这个字段可以在清洗时根据字段名推断比如“作者”字段里带“译”字的归为译者。rating和rating_count建议在导入前就转成数值类型否则 Neo4j 里做排序时会按字符串比较出现“9.0 比 10.0 大”的玄学问题。2.3 导入 Neo4j 并建索引CSV 准备好之后用 Cypher 的 LOAD CSV 导入。下面这段是导入节点和关系的核心语句。// 导入 Book 节点 LOAD CSV WITH HEADERS FROM file:///nodes_Book.csv AS row CREATE (:Book { id: row.id, title: row.title, isbn: row.isbn, pub_year: toInteger(row.pub_year), rating: toFloat(row.rating), rating_count: toInteger(row.rating_count) }); // 导入 Person 节点 LOAD CSV WITH HEADERS FROM file:///nodes_Person.csv AS row CREATE (:Person {id: row.id, name: row.name}); // 导入关系以 WROTE 为例 LOAD CSV WITH HEADERS FROM file:///relationships.csv AS row WITH row WHERE row.rel_type WROTE MATCH (p:Person {id: row.start_id}) MATCH (b:Book {id: row.end_id}) CREATE (p)-[:WROTE]-(b); // 建索引加速后续查询 CREATE INDEX book_title_idx IF NOT EXISTS FOR (b:Book) ON (b.title); CREATE INDEX person_name_idx IF NOT EXISTS FOR (p:Person) ON (p.name); CREATE INDEX movie_title_idx IF NOT EXISTS FOR (m:Movie) ON (m.title);导入顺序很重要先导节点再导关系否则 MATCH 不到。索引建议在导入完成后建数据量大时先建索引再导入反而慢。toInteger和toFloat是必须的CSV 读进来默认全是字符串。如果你的数据超过十万条LOAD CSV 会比较吃力常见做法是改用 neo4j-admin import 或者用 Python 的 py2neo 批量写入。3. 问答层怎么实现从自然语言到 Cypher 查询的映射3.1 意图识别和槽位填充的轻量方案问答系统的核心是把“刘慈欣写了哪些书”翻译成MATCH (p:Person {name:刘慈欣})-[:WROTE]-(b:Book) RETURN b.title。完整做法一般分两步意图识别和槽位填充。意图识别判断用户问的是哪类问题比如“某人作品”“某书作者”“某导演电影”“评分最高”“某年出版”。槽位填充抽出问题里的实体和条件比如人名、书名、年份、评分阈值。不依赖大模型的情况下用规则加词典就能覆盖大部分场景。下面是一个基于正则和关键词的意图解析器。import re # 意图模板意图名 - (正则模式, Cypher 模板) INTENT_PATTERNS [ { name: author_works, pattern: r(.?)(写了|创作了|有哪些|的作品), cypher: MATCH (p:Person {{name:{person}}})-[:WROTE]-(b:Book) RETURN b.title AS 作品, b.rating AS 评分 ORDER BY b.rating DESC }, { name: book_author, pattern: r《?(.?)》?(是谁写的|的作者|作者是谁), cypher: MATCH (p:Person)-[:WROTE]-(b:Book {{title:{book}}}) RETURN p.name AS 作者 }, { name: director_movies, pattern: r(.?)(导演了|导演的|拍了哪些|的电影), cypher: MATCH (p:Person {{name:{person}}})-[:DIRECTED]-(m:Movie) RETURN m.title AS 电影, m.rating AS 评分 ORDER BY m.rating DESC }, { name: top_rated_movie, pattern: r(评分最高|最好看|排名第一).*(电影|片子), cypher: MATCH (m:Movie) RETURN m.title AS 电影, m.rating AS 评分 ORDER BY m.rating DESC LIMIT 5 }, { name: year_books, pattern: r(\d{4})年.*(出版|出了).*(书|作品), cypher: MATCH (b:Book) WHERE b.pub_year {year} RETURN b.title AS 书名, b.rating AS 评分 ORDER BY b.rating DESC LIMIT 10 } ] def parse_question(question): for intent in INTENT_PATTERNS: m re.search(intent[pattern], question) if m: slots {} groups m.groups() if intent[name] author_works: slots[person] groups[0].strip() elif intent[name] book_author: slots[book] groups[0].strip() elif intent[name] director_movies: slots[person] groups[0].strip() elif intent[name] year_books: slots[year] int(groups[0]) cypher intent[cypher].format(**slots) return intent[name], cypher return None, None # 测试 if __name__ __main__: questions [ 刘慈欣写了哪些书, 《三体》是谁写的, 诺兰导演了哪些电影, 评分最高的电影, 2010年出版的书 ] for q in questions: name, cypher parse_question(q) print(f问题: {q}\n意图: {name}\nCypher: {cypher}\n)这段代码用正则做意图匹配每个意图对应一个 Cypher 模板。parse_question遍历所有模板命中后抽取槽位并填充。逻辑说明正则里的(.?)是非贪婪匹配用来抓实体名{{}}是 Python format 的转义因为 Cypher 里也有花括号。参数说明INTENT_PATTERNS列表可以按需扩展每加一个意图就加一条模板槽位名要和 Cypher 模板里的占位符一致。这个方案的边界很明显正则覆盖不了复杂句式比如“刘慈欣写的评分超过 9 分的书有哪些”这种带多重条件的需要叠加条件解析。常见做法是在槽位填充后加一层条件抽取用正则抓“评分超过 X 分”“X 年以后”这类修饰再拼到 Cypher 的 WHERE 子句里。3.2 用 py2neo 执行查询并返回结构化答案解析出 Cypher 之后用 py2neo 连接 Neo4j 执行并格式化结果。from py2neo import Graph class QAEngine: def __init__(self, uribolt://localhost:7687, userneo4j, passwordyour_password): self.graph Graph(uri, auth(user, password)) def answer(self, question): intent, cypher parse_question(question) if not cypher: return {answer: 暂时无法理解这个问题换个说法试试, data: []} try: result self.graph.run(cypher).data() if not result: return {answer: 没有找到相关数据, data: []} return {answer: self._format(intent, result), data: result} except Exception as e: return {answer: f查询出错: {str(e)}, data: []} def _format(self, intent, rows): if intent author_works: titles [r[作品] for r in rows[:5]] return f共找到 {len(rows)} 部作品前五部是{、.join(titles)} elif intent book_author: return f作者是{rows[0][作者]} elif intent director_movies: titles [r[电影] for r in rows[:5]] return f共找到 {len(rows)} 部电影前五部是{、.join(titles)} elif intent top_rated_movie: return f评分最高的电影是《{rows[0][电影]}》评分 {rows[0][评分]} elif intent year_books: return f共找到 {len(rows)} 本书评分最高的是《{rows[0][书名]}》 return str(rows) if __name__ __main__: engine QAEngine(passwordyour_password) print(engine.answer(刘慈欣写了哪些书)) print(engine.answer(评分最高的电影))QAEngine把连接、查询、格式化封装在一起。answer方法先解析问题拿不到 Cypher 就返回兜底话术查询异常时捕获并返回错误信息避免整个服务挂掉。_format按意图类型组织自然语言回答这里只做了简单拼接实际项目里可以做得更细比如加上评分、年份等附加信息。参数上uri默认是本地 Neo4j 的 bolt 端口如果你改过配置要对应调整。password不要硬编码在代码里常见做法是读环境变量或者配置文件。查询结果用.data()转成字典列表方便后续处理。4. 避坑与排查这套系统最容易翻车的五个地方4.1 中文实体名匹配不上现象问“刘慈欣写了哪些书”返回“没有找到相关数据”但数据库里明明有刘慈欣的节点。原因豆瓣数据里人名可能带空格、全角括号或者别名比如“刘慈欣科幻作家”而用户输入的是“刘慈欣”精确匹配失败。解决在 Cypher 里用CONTAINS或者~做模糊匹配比如MATCH (p:Person) WHERE p.name CONTAINS 刘慈欣。更好的做法是在导入时做一次名称归一化去掉括号、空格、别名存一个name_normalized字段专门用于匹配。4.2 LOAD CSV 导入时中文乱码现象导入后节点属性里的中文变成问号或者乱码。原因CSV 文件编码不是 UTF-8或者 Neo4j 的dbms.import.csv.legacy_quote_escaping配置和文件格式不匹配。解决导出 CSV 时显式指定encodingutf-8导入前用file -i nodes_Book.csv确认编码。如果还是乱码检查 Neo4j 配置文件里的dbms.default_csv_encoding确保是 UTF-8。Windows 环境下尤其容易出这个问题因为 Excel 默认存成 GBK。4.3 关系导入时 MATCH 不到节点现象导入关系的 Cypher 执行成功但返回 0 行关系没建上。原因节点导入时 ID 生成规则和关系文件里的 ID 不一致比如节点用了book_123关系里写的是123。解决统一 ID 生成规则节点和关系用同一个函数生成 ID。导入前先跑一句MATCH (n) RETURN count(n)确认节点数再跑MATCH ()-[r]-() RETURN count(r)确认关系数。如果关系数是 0逐条检查关系文件里的 start_id 和 end_id 能不能在节点文件里找到。4.4 问答返回结果顺序不稳定现象问“评分最高的电影”每次返回的电影不一样。原因Cypher 里ORDER BY的字段有重复值比如两部电影评分都是 9.0Neo4j 不保证稳定排序。解决在ORDER BY里加第二排序键比如ORDER BY m.rating DESC, m.rating_count DESC用评分人数做 tie-breaker。如果还不行再加m.title做最终排序。这个坑在演示的时候特别致命因为每次刷新结果都变看起来像 bug。4.5 正则意图匹配误触发现象问“这本书的作者是谁”被author_works意图匹配走了返回一堆书。原因正则模式(.?)(写了|创作了|有哪些|的作品)里的的作品太宽泛把“的作者”也部分匹配了。解决调整正则顺序把更具体的意图放前面比如book_author放在author_works前面。同时给模式加锚点或者边界比如^(.?)的作者限定开头。测试时准备一组边界问题每次改正则都跑一遍回归。5. 进阶技巧把问答准确率从能用推到好用5.1 用同义词词典扩展实体匹配规则匹配最大的问题是用户不会按你预设的说法提问。问“刘慈欣的作品”能匹配问“刘慈欣写的书”也能匹配但问“刘慈欣有啥书”就挂了。解决办法是维护一个同义词词典把“作品”“书”“著作”“写的”都映射到同一个意图。SYNONYMS { 作品: [书, 著作, 写的, 创作], 导演: [执导, 拍的, 导演的], 评分: [打分, 分数, 评价] } def normalize_question(q): for standard, variants in SYNONYMS.items(): for v in variants: q q.replace(v, standard) return q在parse_question之前先调normalize_question把口语化表达统一成标准词。这个词典不用一次写全上线后收集用户问法每周补一批两三轮下来覆盖率就能明显提升。5.2 加一层查询结果校验有些查询语法上没问题但语义上不对。比如问“刘慈欣导演了哪些电影”如果数据库里刘慈欣没有导演关系会返回空结果用户以为系统坏了。更好的做法是在返回空结果时给一个解释性回答比如“刘慈欣在数据库中没有导演记录他作为作者有 15 部作品要看吗”实现上可以在answer方法里加一个 fallback 逻辑主查询返回空时尝试用同一个实体查其他关系类型给用户一个替代选项。这个技巧在演示时特别加分因为用户能感觉到系统“知道自己在干什么”。5.3 用 EXPLAIN 检查 Cypher 性能数据量上来之后有些查询会变慢。Neo4j 提供了EXPLAIN和PROFILE两个命令前者看执行计划后者看实际耗时。EXPLAIN MATCH (p:Person {name:刘慈欣})-[:WROTE]-(b:Book) RETURN b.title ORDER BY b.rating DESC;如果执行计划里出现AllNodesScan说明没走索引需要在Person.name上建索引。如果出现CartesianProduct说明查询里有没连上的模式需要检查关系方向或者加 WHERE 条件。我一般会在开发阶段对每个意图的 Cypher 都跑一遍 PROFILE把db hits超过一万的查询挑出来优化。5.4 一个容易被忽略的细节时间字段的存储格式豆瓣数据里的出版年份和电影上映年份原始格式可能是“2010-01-01”或者“2010年1月”直接存字符串会导致范围查询失效。我一般会在清洗阶段统一转成整数年份存到pub_year和release_year字段同时保留一个pub_date字符串字段用于展示。这样问“2010 年以后出版的书”时Cypher 里直接WHERE b.pub_year 2010就行不用做字符串解析。这套系统我从头搭过两遍第一遍卡在数据清洗上第二遍卡在问答映射上。最大的教训是不要一上来就追求覆盖所有问法先把三五个核心意图跑通把数据质量做扎实再逐步加意图。知识图谱问答的瓶颈从来不是图数据库本身而是你对数据的理解够不够细。希望帮到你。本文还有配套的精品资源点击获取
