简介基于知识图谱的书籍推荐问答系统是一份面向高校软件工程课程的高级大作业完整项目包。系统整合了网络爬虫、知识图谱构建、自然语言处理与Web可视化展示模块结合Flask后端与Neo4j图数据库实现了书籍搜索、推荐和智能问答功能适合正在学习Python、知识图谱或NLP技术的开发者作为综合实战参考。压缩包共415个文件大小约19.72MB其中Py文件承载后端逻辑与爬虫脚本js/css文件支撑前端交互界面jpg/gif图片与json等资源用于数据展示与配置项目结构清晰、开箱可读。目前已有75人学习下载。通过该包可获取完整项目源码、LTP分词与命名实体识别处理流程、知识图谱可视化页面及配套项目说明文档有助于理解从数据采集到图谱问答的系统化实现思路。1. 从豆瓣数据到知识图谱这个问答系统的整体拆解这个项目最花时间的不是推荐算法而是“让图数据库和自然语言对齐”。系统围绕豆瓣书单构建知识图谱清洗后的书籍、作者、出版社、标签被写入 Neo4j后端用 Python Flask 提供 HTTP 接口问答模块用 LTP 做分词、词性标注和命名实体识别再把“刘慈欣有哪些代表作”这类中文问句翻译成 Cypher 查询。它更像一条“结构化爬虫 图存储 NLP 映射”的完整链路而不是纯推荐算法实验。适合已经有爬虫和 Flask 基础、但没碰过图数据库的人。新手可以把每个模块当作独立示例跑通老手值得关注的是实体归一化、关系方向统一、图数据序列化这些容易翻车的地方。2. 爬虫数据清洗与实体关系建模为了 Neo4j 存储而做的准备2.1 实体与关系不要把数据库表结构直接搬进图数据库很多课程作业的第一步是建数据库表但图数据库建模应该先画实体和关系。豆瓣上“书”和“作者”是多对多“书”和“标签”也是多对多如果用 MySQL 设计需要 book、author、book_author、book_tag 至少四张表回答“刘慈欣有哪些代表作”要 join 三次。Neo4j 里只需要沿WROTE关系走一步这也是知识图谱推荐问答系统把数据放在图数据库里的主要原因。实体标签主要属性来源Bookbook_id、title、rating、rating_num、intro、cover图书列表页和详情页Authorauthor_id、name、intro作者主页 / 图书贡献者Publisherpublisher_id、name图书出版信息Tagname图书标签云关系我建议统一写成主动式(Author)-[:WROTE]-(Book)、(Book)-[:HAS_TAG]-(Tag)、(Publisher)-[:PUBLISHED]-(Book)。注意WROTE和AUTHORED_BY都有人用但方向必须统一不然问答模板里是MATCH (a:Author)-[:WROTE]-(b:Book)后台导入时写的却是MERGE (b)-[:AUTHORED_BY]-(a)那么前端图谱里会出现“书写了作者”的假关系而且很难查出来。建议在项目文档里先用一句话固定关系方向和命名再写代码。2.2 爬虫数据清洗要点从列表页到可入库字段豆瓣列表页的每个.subject-item块里标题、评分、出版信息在不同节点。课程项目一般先从列表页抓主要字段缺啥再补详情页避免一开始就全量采集。# crawl_books.py - 课程项目里的简化版列表页采集 import requests from bs4 import BeautifulSoup def parse_book_item(item): title item.select_one(h2 a) rating item.select_one(.rating_nums) pub item.select_one(.pub) if not title: return None return { title: title.text.strip(), rating: float(rating.text.strip()) if rating else 0.0, pub: pub.text.strip() if pub else } def fetch_subject_list(url, sessionNone): session session or requests.Session() resp session.get(url, headers{User-Agent: Mozilla/5.0}) resp.raise_for_status() soup BeautifulSoup(resp.text, html.parser) rows [] for item in soup.select(.subject-item): row parse_book_item(item) if row: rows.append(row) return rows逻辑说明select_one是 BeautifulSoup 里最常用的选择方法只取第一个匹配节点拿不到评分时返回0.0保证后续写入 Neo4j 的字段类型一致不需要在 Cypher 里做空值判断。session.get带上User-Agent只是最基础的请求头真正跑全量还要加限速、重试和随机休眠否则 IP 很快被限流。参数说明url是豆瓣某个分类的列表页比如https://book.douban.com/top250session参数允许外部传入带 cookie 的会话方便翻页时复用连接。豆瓣的pub字段通常是[美] 卡尔·纽波特 / 张宝元 / 北京联合出版公司 / 2021-4这种格式直接用split(/)拆分def split_pub(pub_text): # 豆瓣出版信息一般以 / 分隔作者、译者、出版社、年份 parts [p.strip() for p in pub_text.split(/)] if len(parts) 3: return {author_text: pub_text, publisher: , year: } return { author_text: /.join(parts[:-2]), publisher: parts[-2], year: parts[-1] }这里把倒数第二项当出版社、最后一项当出版年是豆瓣格式的常见规律。但“2021-4”里还包含月份需要后续用正则re.search(r(19|20)\d{2}-\d{1,2}, parts[-1])抽取不能直接把整个字符串存成 year。我一般会在这一层保留author_text的原始内容因为作者可能有多人“刘慈欣 / 王晋康”不能简单塞进一个字段等写入图时再用UNWIND split(author_text, /)展开成多个作者节点。如果你在这里过早拆成数组后面和 LTP 实体名做匹配时反而要多一次 join。2.3 批量写入 Neo4j用 MERGE 而不是 CREATE数据清洗完就该落到 Neo4j。第一次写时很容易用CREATE跑完发现一个“刘慈欣”有几十个节点因为每本书的贡献列表里都重复出现同一作者。正确做法是先为关键属性建唯一约束CREATE CONSTRAINT book_id_unique IF NOT EXISTS ON (b:Book) ASSERT b.book_id IS UNIQUE; CREATE CONSTRAINT author_name_unique IF NOT EXISTS ON (a:Author) ASSERT a.name IS UNIQUE;唯一约束相当于图数据库里的唯一索引它保证后续MERGE按book_id或name去重时不会产生重复实体。注意IF NOT EXISTS是 Neo4j 4.x 的写法3.x 里直接执行CREATE CONSTRAINT ON ...即可重复执行会报错。然后在 Python 驱动里用MERGE写入from neo4j import GraphDatabase driver GraphDatabase.driver(bolt://localhost:7687, auth(neo4j, 123456)) def save_book(tx, book): # MERGE 按唯一属性去重避免重复创建作者节点 tx.run( MERGE (b:Book {book_id: $book_id}) SET b.title $title, b.rating $rating WITH b UNWIND $authors AS author_name MERGE (a:Author {name: author_name}) MERGE (a)-[:WROTE]-(b) , book_idbook[book_id], titlebook[title], ratingbook[rating], authorsbook[authors] )逻辑说明UNWIND $authors会把 Python 列表拆成一行一个作者然后逐个MERGE Author再创建WROTE关系。这样做的好处是一条 Cypher 就能写完一本书和它的所有作者但坑在于如果authors是空列表UNWIND []会产生零行导致后面的MERGE以及这条事务里的Book插入全部被跳过。所以调用前要保证authors至少是[未知作者]或者把书节点单独插入。参数说明book_id用豆瓣自身的编号能天然去重rating在 Python 里已经是 float驱动会按 Float 传给 Neo4j不需要在 Cypher 里再toFloat。写入完成后我习惯先在 Neo4j 浏览器跑一个验证查询MATCH (a:Author {name:刘慈欣})-[:WROTE]-(b:Book) RETURN count(b)如果数量明显大于真实书目多半是作者名带了“著”“主编”等后缀需要回第二步做清洗。批量导入数据时逐本书session.execute_write(save_book, book)即可如果一次把几千本书塞进同一个事务中途任何一条数据出问题都会全部回滚不好定位。3. LTP 自然语言处理与查询意图识别分词、词性标注与命名实体识别实战3.1 为什么用 LTP 而不是裸正则问答模块的难点在于问题变体多比如“刘慈欣有哪些代表作”“推荐几本刘慈欣的书”“刘慈欣写得最好的是哪本”关键词不完全一样但实体都是“刘慈欣”。用正则写匹配规则会越写越长最后变成一堆in判断。LTP语言技术平台提供中文分词、词性标注、命名实体识别能先把“刘慈欣”作为一个整体切出来后面只需要拿实体名去图里查。相比 jieba 只做分词LTP 的好处是自带的 NER 模型可以标出人名、地名、机构名和知识图谱实体对齐更自然相比加载完整 spaCy 中文模型LTP 在课程作业场景下更轻量模型文件更小。实际项目中LTP 有新旧两个版本旧版用pyltp加载ltp_data里的模型新版用ltp包。课程作业大多是旧版代码也以这个风格呈现# nlp_utils.py from pyltp import Segmentor, Postagger, NamedEntityRecognizer MODEL_DIR ./ltp_data segmentor Segmentor() segmentor.load(MODEL_DIR /cws.model) postagger Postagger() postagger.load(MODEL_DIR /pos.model) ner NamedEntityRecognizer() ner.load(MODEL_DIR /ner.model) def analyze_question(text): words list(segmentor.segment(text)) postags list(postagger.postag(words)) netags list(ner.recognize(words, postags)) return words, postags, netags逻辑说明三个模型在启动时一次性加载到全局变量问答请求直接复用不需要每次重新加载否则一次请求要等几十毫秒甚至更久。但Segmentor、Postagger这些实例不是线程安全的Flask debug 模式默认单线程问题不大生产或多线程跑会偶发异常。常见做法是用threading.local()包一层让每个线程拿到自己的 LTP 实例答辩阶段只要提到这个坑老师一般不会再深挖。参数说明MODEL_DIR指向解压后的 LTP 模型目录里面必须有cws.model、pos.model、ner.model三个文件如果加载时报错提示找不到模型优先检查这个路径而不是 Python 代码。LTP 输出结果大致如下输入wordsNER 标签刘慈欣有哪些代表作[刘慈欣, 有, 哪些, 代表作][Nh, O, O, O]推荐几本东野圭吾的悬疑小说[推荐, 几本, 东野圭吾, 的, 悬疑, 小说][O, O, Nh, O, O, O]三体这本书怎么样[三体, 这, 本, 书, 怎么样][O, O, O, O, O]注意 LTP 老版本的人名标签是Nh机构是Ni地名是Ns书名一般不会出现在 NER 结果里因为“三体”在训练语料里不一定被标成作品名。所以analyze_question只是提供候选实体最终是否采用还要看图数据库里有没有同名节点。3.2 意图识别把 NLP 输出映射到 Cypher 模板拿到words、postags、netags之后需要决定用什么 Cypher 查询。我一般维护一个优先级列表先看有没有Nh人名再看有没有“推荐”“有哪些”这类意图词最后兜底用书名模糊匹配。def build_cypher(words, postags, netags, entity_map): # 先识别命名实体再作为 Author 查询条件 entities [w for w, n in zip(words, netags) if n ! O] if entities: name normalize_entity(entities[0], entity_map) return { query: MATCH p(a:Author {name:$name})-[:WROTE]-(b:Book) RETURN p LIMIT 20, params: {name: name} } if 推荐 in words or 有哪些 in .join(words): return { query: MATCH p(b:Book)-[:HAS_TAG]-(t:Tag) WHERE t.name $tag RETURN p LIMIT 20, params: {tag: extract_tag(words)} } return { query: MATCH p(b:Book) WHERE b.title CONTAINS $q RETURN p LIMIT 20, params: {q: .join(words)} }逻辑说明这个函数是问答系统的核心但它故意写得简单重点在于把“NLP 结果”和“图查询”解耦。比如entities[0]可能是地名“北京”直接拿去MATCH (a:Author {name:北京})会返回空所以更稳的做法是先对entities[0]在 Neo4j 里同时查Author和Tag两种标签哪个命中用哪个上面为了可读性没有写这类兼容。推荐 in words是一个迭代查找实际运行没问题如果你担心分词把“推荐”拆成“推/荐”可以写成any(w in (推荐, 有没有, 哪些) for w in words)。参数说明entity_map是从 Neo4j 作者表加载出来的实体集合用于把 LTP 结果归一化到图谱里真实存在的名字extract_tag(words)是另一个辅助函数我用它去掉停用词后把剩余名词当成标签候选。注意模板里LIMIT 20限制的是返回路径条数不是节点数实际前端渲染的节点可能比 20 多业务中应把这个上限做成接口参数。3.3 实体归一化解决“刘慈欣”与“刘慈欣 著”不一致从 LTP 得到的人名经常和图数据库里的Author.name不一致。列表页抓来的作者名带“著”“译”是常态也有一些书把作者写作“刘慈欣 等著”。因此需要一张实体字典在 Flask 启动时从 Neo4j 加载所有作者名def load_entity_dict(driver): with driver.session() as session: result session.run(MATCH (a:Author) RETURN a.name AS name) return {row[name].strip() for row in result} def normalize_entity(raw, name_set): # 先用精确匹配再尝试包含匹配解决“刘慈欣 著”这类后缀问题 if raw in name_set: return raw for name in name_set: if raw in name or name in raw: return name return raw逻辑说明raw in name能把“刘慈欣”匹配到“刘慈欣 著”name in raw能把“刘慈欣 著”匹配到“刘慈欣”。看起来简单但要注意name_set是集合遍历时顺序不确定如果多个名字都包含“刘慈欣”返回哪个会有随机性。解决方案是给实体字典加一个打分先按精确命中再按长度倒序取最长匹配能避免“刘慈欣”误匹配到“刘慈欣 编”而不是“刘慈欣”。课程作业里通常数据量不大直接遍历没问题数据量大时可以把 key 做标准化后再精确查一遍def build_key(name): # 去掉空格和常见角色后缀只保留作者核心名 return name.replace( , ).replace(著, ).replace(编, ) entity_dict {build_key(name): name for name in author_names} def normalize_fast(raw): return entity_dict.get(build_key(raw), raw)参数说明build_key里的替换规则要根据你清洗后的实际数据调整比如还有“美”“ [美]”这类前缀需要先用正则去掉方括号和括号内容。这个函数放在 NLP 模块还是数据清洗模块都可以但一定要和build_cypher使用同一个entity_map不能一个用旧的一个用新的。4. Flask API 与 Neo4j 查询逻辑从 Cypher 到前端可视化4.1 路由设计不要把查询逻辑写在视图函数里项目静态文件里有bootstrap.min.css、layui.css、wiki.css说明前端是后台管理界面风格交互以搜索框、结果列表和图谱展示为主。后端接口我习惯拆成这样端点方法参数返回/api/askPOSTquestion问答结果、图谱节点和边/api/searchGETq、limit书籍列表/api/recommendGETbook_id推荐书籍列表/api/entityGETname作者或实体详情每个路由只做参数校验、调用处理函数、序列化响应不在视图函数里写复杂 NLP 和 Cypher# routes.py from flask import Blueprint, request, jsonify api Blueprint(api, __name__) api.post(/api/ask) def ask(): body request.get_json(silentTrue) or {} question body.get(question, ) if not question.strip(): return jsonify({error: question is empty}), 400 words, postags, netags analyze_question(question) candidate build_cypher(words, postags, netags, entity_dict) with driver.session() as session: records session.run(candidate[query], candidate[params]).data() # 把 Cypher 返回的路径转成 ECharts 可用的图结构 graph records_to_graph(records) return jsonify({ question: question, nodes: graph[nodes], edges: graph[edges] })逻辑说明api.post是 Flask 2.0 以后的简写课程项目如果还在用 Flask 1.x要写成api.route(/api/ask, methods[POST])。request.get_json(silentTrue) or {}能同时处理前端没传Content-Type、传了空 body、传了非法 JSON 三种情况避免直接抛 400。参数说明entity_dict是启动时从 Neo4j 加载的实体字典不能放在路由函数内重复加载session.run(...).data()会把每条记录转换成 dict但Node和Relationship还需要后续序列化。4.2 Cypher 图数据序列化把节点和关系拆成前端能用的 JSON前端画知识图谱常用 ECharts 的 graph 类型它要求的是{ nodes: [...], edges: [...] }而后端 Cypher 返回的是Node和Relationship对象直接jsonify会报Object of type Node is not JSON serializable。因此需要一层转换def records_to_graph(records, path_keyp): nodes, edges, seen_edges {}, [], set() for rec in records: path rec.get(path_key) if path is None: continue # 先收集路径上的所有节点Book、Author、Tag 都包含 for node in path.nodes: nid str(node.element_id) if nid not in nodes: labels list(node.labels) nodes[nid] { id: nid, label: labels[0] if labels else NODE, title: node.get(title) or node.get(name) or entity, } # 再收集关系用三元组去重避免多条路径返回同一条边 for rel in path.relationships: start, end rel.start_node, rel.end_node edge_key (str(start.element_id), str(end.element_id), rel.type) if edge_key not in seen_edges: seen_edges.add(edge_key) edges.append({ source: str(start.element_id), target: str(end.element_id), type: rel.type }) return {nodes: list(nodes.values()), edges: edges}逻辑说明path是 Cypher 里返回的路径变量p遍历path.nodes可以拿到路径上所有节点遍历path.relationships可以拿到所有关系。element_id是 Neo4j 4.x 的新 API如果项目还在 Neo4j 3.5需要改成node.id。一个更稳妥的方案是在节点属性里额外维护node_id字符串前端用这个属性做 id驱动版本升级不影响。参数说明path_key默认是p如果你的 Cypher 写的是RETURN path这里改成path即可node.get(title)只对 Book 节点有效Author 节点要取name。如果用图查询语句返回整条路径而不是普通表格常见写法MATCH p(a:Author {name:$name})-[:WROTE]-(b:Book)-[:HAS_TAG]-(t:Tag) RETURN p LIMIT 50注意LIMIT 50限制的是路径条数不是节点数。一条路径可能包含 3 个节点和 2 条关系前端实际渲染的节点数可能是 50 到 150 个如果数据量大要用collect聚合后再UNWIND去重或者直接把限制调小。在课程答辩时50 条路径已经足够展示图谱效果不会让浏览器卡死。4.3 推荐接口用共享标签或共同作者做基于图的推荐“推荐类似书”在 Neo4j 里非常直接找与当前书共享HAS_TAG标签的其他书按重合标签数排序。Cypher 查询如下MATCH (b:Book {book_id:$book_id})-[:HAS_TAG]-(t:Tag)-[:HAS_TAG]-(cand:Book) WHERE cand.book_id $book_id WITH cand, count(t) AS shared_tags, max(cand.rating_num) AS rating_num WHERE rating_num $min_raters ORDER BY shared_tags DESC, cand.rating DESC RETURN cand.title AS title, cand.rating AS rating, shared_tags LIMIT 10逻辑说明先用当前书的标签找到所有候补书count(t)统计的是“当前书和候补书共同拥有的标签数”这个值越高推荐相关性越强。用cand.book_id $book_id排除自己用rating_num $min_raters过滤掉只有几条评价的书避免冷门书因为蹭了热门标签而上榜。参数说明rating_num是豆瓣的评分人数不是评分如果列表页没有抓这个字段就直接去掉WHERE rating_num不要让它影响查询结果。Cypher 里不能直接给max(cand.rating_num)起别名后在同一个WHERE中使用所以上面先WITH了一次这个细节在 Neo4j 里很容易踩坑。对应 Flask 接口api.get(/api/recommend) def recommend(): book_id request.args.get(book_id) if not book_id: return jsonify({error: book_id is required}), 400 # 如果 Neo4j 里 book_id 是字符串这行可以去掉否则必须转 int try: book_id int(book_id) except ValueError: return jsonify({error: book_id must be int}), 400 query MATCH (b:Book {book_id:$book_id})-[:HAS_TAG]-(t:Tag)-[:HAS_TAG]-(cand:Book) WHERE cand.book_id $book_id WITH cand, count(t) AS shared_tags, max(cand.rating_num) AS rating_num WHERE rating_num $min_raters ORDER BY shared_tags DESC, cand.rating DESC RETURN cand.title AS title, cand.rating AS rating, shared_tags LIMIT 10 with driver.session() as session: rows session.run(query, book_idbook_id, min_raters100).data() return jsonify({recommendations: rows})这里有个典型的类型坑request.args.get(book_id)拿到的永远是字符串而 Neo4j 中book_id属性可能是整数直接传字符串会查不到。解决方式是在 Python 里int(book_id)或 Cypher 里toInteger($book_id)。如果你在爬虫阶段把book_id设计成字符串很多豆瓣编号以 0 开头就没有这个问题但如果当时用int()转了整数必须统一。我一般建议爬虫阶段不要轻易改变豆瓣原始编号的类型因为后面做增量更新时还要靠它去重。4.4 前端可视化最少代码回答“书籍推荐问答系统”展示效果前端最少需要这样一个 ECharts 调用// graph-view.js async function ask() { const q document.getElementById(q).value; const res await fetch(/api/ask, { method: POST, headers: {Content-Type: application/json}, body: JSON.stringify({question: q}) }); const data await res.json(); // 后端的空结果也会带 nodes: []这里避免 setOption 报错 if (!data || !data.nodes) return; myChart.setOption({ series: [{ type: graph, layout: force, roam: true, label: {show: true, fontSize: 12}, data: data.nodes, edges: data.edges }] }); }逻辑说明roam: true允许用户在展示时拖拽、缩放这对图谱类页面几乎是必需的。layout: force适合 100 个节点以内的小图节点多了会一直在布局抖动课程项目不会有这个问题如果节点超过 80建议改成circular否则演示现场容易卡顿。参数说明data.nodes直接使用后端records_to_graph生成的id、label、title字段ECharts 的edges要求source和target与nodes中的id对应如果后端返回的是from/to需要做字段映射。另外前端项目里同时引用了layui.js和jquery.js的话要注意$和layui.$的冲突最好只选一个做事件绑定。5. 部署、调试与效果验证用日志和测试用例验收问答系统5.1 本地快速启动与验收清单想快速验证这套 Python 知识图谱问答系统先启动 Neo4j 社区版容器注意不需要企业版授权# 浏览器控制台端口7474驱动端口7687 docker run -d --name neo4j-book \ -p 7474:7474 -p 7687:7687 \ -e NEO4J_AUTHneo4j/123456 \ neo4j:4.4-community export FLASK_ENVdevelopment python app.py # 另开一个终端验证问答接口 curl -X POST http://localhost:5000/api/ask \ -H Content-Type: application/json \ -d {question:刘慈欣有哪些代表作} | python -m json.tool逻辑说明NEO4J_AUTHneo4j/123456会在容器首次启动时初始化数据库密码如果容器后面被停止再启动这个变量不会再生效要重置密码就去浏览器控制台改。FLASK_ENVdevelopment让 Flask 启动 reloader方便本地改代码即时生效但注意首次启动时 LTP 模型加载需要几秒钟reloader 会自动重启一次不要以为进程崩溃了。curl 命令用了-X POST实际上-d参数会自动让请求变成 POST保留-X更多是提醒自己这是 POST 接口。现象排查点问题包含人名但返回空Neo4j 浏览器执行MATCH (a:Author {name:刘慈欣}) RETURN a看名称是否有后缀前端图谱不渲染浏览器开发者工具看/api/ask响应里是否有nodes/edges字段推荐结果重复Cypher 是否排除了book_id本身UNWIND标签列表是否有空值模型加载报错LTPMODEL_DIR下是否同时存在cws.model、pos.model、ner.model最后一个建议把第 3 章的analyze_question单独跑一遍再启动 Flask用python -c from nlp_utils import analyze_question; print(analyze_question(测试一下))确认模型路径没问题接口写完后再用 pytest 或 curl 固定 3 条测试用例比如“刘慈欣有哪些代表作”“推荐几本推理小说”“三体”每轮改完代码都跑一遍比在浏览器里反复点提交效率高得多。如果自己构建的图数据超过十万节点再用 Neo4j admin import 工具做批量导入课程项目数据量小直接在 Python 驱动里按事务写入就能保持代码可读性。本文还有配套的精品资源点击获取