简介：这是一套基于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_count；Movie 要有 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', encoding='utf-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 = f"book_{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 = f"person_{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 = f"pub_{rec['publisher']}" nodes['Publisher'][pid] = {'name': rec['publisher']} rels.append(('Publisher', pid, 'PUBLISHED', 'Book', bid)) # 标签 for tag in rec.get('tags', []): tid = f"tag_{tag}" nodes['Tag'][tid] = {'name': tag} rels.append(('Book', bid, 'HAS_TAG', 'Tag', tid)) elif rec.get('type') == 'movie': mid = f"movie_{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 = f"person_{d['id']}" nodes['Person'][did] = {'name': d['name']} rels.append(('Person', did, 'DIRECTED', 'Movie', mid)) for a in rec.get('actors', []): aid = f"person_{a['id']}" nodes['Person'][aid] = {'name': a['name']} rels.append(('Person', aid, 'ACTED_IN', 'Movie', mid)) for g in rec.get('genres', []): gid = f"genre_{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_ok=True) 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='', encoding='utf-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='', encoding='utf-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, uri='bolt://localhost:7687', user='neo4j', password='your_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(password='your_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 时显式指定encoding='utf-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.0，Neo4j 不保证稳定排序。
解决：在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就行，不用做字符串解析。
这套系统我从头搭过两遍，第一遍卡在数据清洗上，第二遍卡在问答映射上。最大的教训是：不要一上来就追求覆盖所有问法，先把三五个核心意图跑通，把数据质量做扎实，再逐步加意图。知识图谱问答的瓶颈从来不是图数据库本身，而是你对数据的理解够不够细。希望帮到你。
本文还有配套的精品资源，点击获取
