当前位置：首页 > news >正文

一个知识问答系统，用户的选择决定接下来出现的问题，且下一个问题的呈现取决于前面几个问题的回答

news 来源：原创 2024/9/29 15:31:12

一个知识问答系统，用户的选择决定接下来出现的问题，且下一个问题的呈现取决于前面几个问题的回答，我们需要设计一个更复杂的图结构来表示这些关系。

设计图结构

Question节点：表示问题。
Answer节点：表示答案。
HAS_ANSWER关系：链接问题和答案。
DEPENDS_ON关系：链接下一个问题与依赖的答案。

图数据库架构

Question节点：
- id：问题ID。
- text：问题文本。
- is_multiple_choice：是否为多选题。
Answer节点：
- id：答案ID。
- text：答案文本。
- is_correct：是否为正确答案（可选）。
HAS_ANSWER关系：
- 从Question到Answer。
DEPENDS_ON关系：
- 从Question到Answer，表示下一个问题依赖于多个前面问题的答案。

实现步骤

1. 安装和配置依赖

首先安装Neo4j驱动：

pip install neo4j

2. 初始化Neo4j连接

from neo4j import GraphDatabaseclass Neo4jConnection:def __init__(self, uri, user, pwd):self.__uri = uriself.__user = userself.__password = pwdself.__driver = Nonetry:self.__driver = GraphDatabase.driver(self.__uri, auth=(self.__user, self.__password))except Exception as e:print("Failed to create the driver:", e)def close(self):if self.__driver is not None:self.__driver.close()def query(self, query, parameters=None, db=None):assert self.__driver is not None, "Driver not initialized!"session = Noneresponse = Nonetry:session = self.__driver.session(database=db) if db is not None else self.__driver.session()response = list(session.run(query, parameters))except Exception as e:print("Query failed:", e)finally:if session is not None:session.close()return response# Initialize connection
conn = Neo4jConnection(uri="bolt://localhost:7687", user="neo4j", pwd="password")

3. 定义数据模型

from pydantic import BaseModel
from typing import Listclass Answer(BaseModel):id: strtext: stris_correct: bool = Falseclass Question(BaseModel):id: strtext: stris_multiple_choice: boolanswers: List[Answer]class Dependency(BaseModel):question_id: strdepends_on: List[str]  # List of answer IDs that the question depends on

4. FastAPI路由和业务逻辑

from fastapi import FastAPI, HTTPException
from typing import Listapp = FastAPI()@app.post("/questions/")
def create_question(question: Question):query = """CREATE (q:Question {id: $id, text: $text, is_multiple_choice: $is_multiple_choice})WITH qUNWIND $answers AS answerCREATE (a:Answer {id: answer.id, text: answer.text, is_correct: answer.is_correct})CREATE (q)-[:HAS_ANSWER]->(a)"""conn.query(query, parameters={"id": question.id, "text": question.text, "is_multiple_choice": question.is_multiple_choice, "answers": [a.dict() for a in question.answers]})return {"status": "Question created"}@app.post("/dependencies/")
def create_dependency(dependency: Dependency):for answer_id in dependency.depends_on:query = """MATCH (q:Question {id: $question_id}), (a:Answer {id: $answer_id})CREATE (q)-[:DEPENDS_ON]->(a)"""conn.query(query, parameters={"question_id": dependency.question_id, "answer_id": answer_id})return {"status": "Dependency created"}@app.get("/questions/{question_id}", response_model=Question)
def get_question(question_id: str):query = """MATCH (q:Question {id: $id})-[:HAS_ANSWER]->(a:Answer)RETURN q, collect(a) as answers"""result = conn.query(query, parameters={"id": question_id})if not result:raise HTTPException(status_code=404, detail="Question not found")record = result[0]question_node = record["q"]answers = [Answer(id=answer["id"], text=answer["text"], is_correct=answer["is_correct"]) for answer in record["answers"]]question = Question(id=question_node["id"], text=question_node["text"], is_multiple_choice=question_node["is_multiple_choice"], answers=answers)return question@app.post("/questions/{question_id}/next/")
def get_next_question(question_id: str, selected_answers: List[str]):query = """MATCH (next_q:Question)-[:DEPENDS_ON]->(a:Answer)WHERE a.id IN $selected_answersRETURN next_q"""result = conn.query(query, parameters={"selected_answers": selected_answers})if not result:raise HTTPException(status_code=404, detail="No dependent questions found")next_question_id = result[0]["next_q"]["id"]return get_question(next_question_id)