昨天寫了一篇文章，使用fastapi直接操作neo4j圖數(shù)據(jù)庫插入數(shù)據(jù)的例子， 本文實(shí)現(xiàn)LLM大模型結(jié)合neo4j圖數(shù)據(jù)庫實(shí)現(xiàn)AI問答功能。

廢話不多說，先上代碼

import gradio as gr
from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel
from langchain_openai import ChatOpenAI
from langchain.prompts import (ChatPromptTemplate,MessagesPlaceholder,SystemMessagePromptTemplate,HumanMessagePromptTemplate,
)
from langchain.chains import LLMChain
from langchain.memory import ConversationBufferMemory
from langchain_community.graphs import Neo4jGraph
from langchain.chains import GraphCypherQAChain
import asyncio
from typing import List
import json# Initialize FastAPI
app = FastAPI()# Initialize Neo4j with timeout
try:graph = Neo4jGraph(url="bolt://localhost:7687",username="neo4j",password="password",database="neo4j",timeout=60  # 60 seconds timeout)
except Exception as e:print(f"Failed to connect to Neo4j: {e}")graph = None# Fallback in-memory storage
job_seekers = []
job_positions = []# Initialize LangChain components
llm = ChatOpenAI(temperature=0.95,model="glm-4-flash",openai_api_key="xxxxxx",openai_api_base="https://open.bigmodel.cn/api/paas/v4/"
)prompt = ChatPromptTemplate(messages=[SystemMessagePromptTemplate.from_template("You are a helpful AI assistant for a recruitment company. You can answer questions about job seekers and available positions."),MessagesPlaceholder(variable_name="chat_history"),HumanMessagePromptTemplate.from_template("{question}")]
)memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
conversation = LLMChain(llm=llm,prompt=prompt,verbose=True,memory=memory
)# Initialize GraphCypherQAChain if Neo4j is available
if graph:graph_qa = GraphCypherQAChain.from_llm(llm,graph=graph,verbose=True)# Define chat function with timeout
async def chat_with_timeout(message, history):try:if graph:neo4j_response = await asyncio.wait_for(asyncio.to_thread(graph_qa.run, message),timeout=10.0  # 10 seconds timeout)return f"Based on our database: {neo4j_response}"else:# Fallback to in-memory dataif "job seekers" in message.lower():return f"Based on our records: We have {len(job_seekers)} job seekers."elif "job positions" in message.lower():return f"Based on our records: We have {len(job_positions)} job positions."else:response = conversation.invoke({"question": message})return response['text']except asyncio.TimeoutError:return "I'm sorry, but the database query took too long. Please try a simpler question or try again later."except Exception as e:print(f"Error in chat function: {e}")response = conversation.invoke({"question": message})return response['text']# # Create Gradio interface
iface = gr.ChatInterface(chat_with_timeout)
#
# # Mount Gradio app to FastAPI
app = gr.mount_gradio_app(app, iface, path="/")# Run the app
if __name__ == "__main__":import uvicornuvicorn.run(app, host="0.0.0.0", port=8000)

還是老規(guī)矩，先AI解釋下，構(gòu)建一個(gè)基于FastAPI和Gradio的聊天應(yīng)用，主要功能如下：

1、初始化FastAPI應(yīng)用和Neo4j圖數(shù)據(jù)庫連接（帶超時(shí)處理）；

2、定義了用于對(duì)話的LangChain組件，包括LLM模型、提示模板及對(duì)話記憶；

3、根據(jù)Neo4j是否可用初始化圖查詢鏈；

4、實(shí)現(xiàn)異步聊天函數(shù)，支持?jǐn)?shù)據(jù)庫查詢數(shù)據(jù)檢索，并處理超時(shí)錯(cuò)誤；

5、使用Gradio創(chuàng)建用戶界面并將應(yīng)用掛載到FastAPI上。

核心關(guān)注graph_qa.run方法，執(zhí)行原理：

1. 自然語言處理：

當(dāng)調(diào)用 graph_qa.run(message) 時(shí)，首先會(huì)將用戶的自然語言查詢（message）傳遞給大語言模型（LLM）。

2. Cypher 查詢生成：

LLM 分析用戶的查詢，并嘗試將其轉(zhuǎn)換為 Cypher 查詢語言。Cypher 是 Neo4j 圖數(shù)據(jù)庫使用的查詢語言。這個(gè)步驟涉及到理解用戶意圖和將其映射到圖數(shù)據(jù)庫的結(jié)構(gòu)上。

3. 數(shù)據(jù)庫查詢：

生成的 Cypher 查詢被發(fā)送到 Neo4j 數(shù)據(jù)庫執(zhí)行。這個(gè)過程涉及到遍歷圖數(shù)據(jù)庫，匹配節(jié)點(diǎn)和關(guān)系，并檢索相關(guān)數(shù)據(jù)。

4. 結(jié)果解釋：

數(shù)據(jù)庫返回查詢結(jié)果后，這些結(jié)果會(huì)被傳回給 LLM。LLM 會(huì)分析這些原始數(shù)據(jù)，理解其含義和上下文。

5. 響應(yīng)生成：

最后，LLM 會(huì)根據(jù)原始查詢和數(shù)據(jù)庫返回的結(jié)果，生成一個(gè)人類可讀的響應(yīng)。這個(gè)響應(yīng)應(yīng)該直接回答用戶的問題，并可能包含從數(shù)據(jù)庫中提取的具體信息。

在上一篇文章中，我已經(jīng)在neo4j插入了一些數(shù)據(jù)，比如張三1的技能。 這里問一下

原文鏈接： 【知識(shí)圖譜】4、LLM大模型結(jié)合neo4j圖數(shù)據(jù)庫實(shí)現(xiàn)AI問答的功能