使用图数据库的检索增强生成（RAG）

本文深入探讨了如何将图数据库（以Neo4j为例）集成到RAG系统中，以增强大语言模型处理复杂关系和多跳推理的能力。文章从图数据库的优势出发，详细介绍了构建知识图谱、生成向量索引、混合检索（向量相似性+图遍历）以及生成答案的完整流程。通过一个产品推荐聊天机器人的实战案例，展示了使用Langchain和OpenAI API实现实体提取、Cypher查询生成、相似物品推荐等关键步骤。同时，文章对比了基于Langchain Agent的对话式方案与纯代码确定性方案的优劣，并提供了常见错误排查与生产环境优化建议，帮助开发者构建更可靠、可解释的AI应用。

技术背景

传统RAG系统依赖向量数据库进行语义相似性检索，但向量检索本质上是近似搜索，难以精确处理实体间的结构化关系和多跳推理问题。例如，在回答“哪些蓝色服装适合成年人？”时，需要同时匹配颜色、类别和年龄组三个维度，且这些维度之间存在逻辑关联。图数据库以节点和边存储数据，天然支持精确查询和关系遍历，能够提供可解释的推理路径。将图数据库与向量检索结合，可以兼顾语义相似性和结构化关系的优势，适用于企业知识管理、推荐系统、智能问答等场景。然而，这种方法也面临数据建模复杂、查询优化困难、混合检索策略设计等挑战。

核心原理

核心原理是将知识图谱作为外部知识源，通过两阶段检索增强LLM生成。第一阶段，利用向量索引对用户查询进行语义相似性搜索，快速定位相关节点（如产品）；第二阶段，基于图结构进行遍历，获取与这些节点关联的上下文（如类别、品牌、颜色等）。这种混合检索策略既利用了向量嵌入的语义理解能力，又发挥了图数据库的关系推理优势。关键设计决策包括：1）实体提取：使用LLM从用户提示中提取结构化实体（如颜色、类别），避免直接生成Cypher查询带来的错误；2）相似度阈值：通过余弦相似度阈值控制检索精度；3）图遍历规则：定义相似物品的判定标准（如同类别且至少一个共同实体）。

架构分析

系统架构分为四层：数据层（Neo4j图数据库存储产品节点和实体节点，以及它们之间的关系）、嵌入层（使用OpenAI Embedding API为节点生成向量，并创建向量索引）、检索层（包含实体提取模块、Cypher查询生成模块、向量相似性搜索模块和图遍历模块）、生成层（将检索结果与用户查询组合，通过LLM生成最终答案）。数据流为：用户查询→实体提取（LLM）→生成Cypher查询→执行查询（结合向量相似度）→获取候选产品→图遍历获取相似物品→LLM生成响应。组件间通过Langchain框架集成，支持两种模式：基于Agent的对话式（灵活但可能产生幻觉）和纯代码确定性（稳定但交互性弱）。

完整实战教程

使用图数据库的检索增强生成

检索增强生成（RAG）是一种强大的技术，它通过从外部知识源检索相关信息来增强大语言模型（LLM）的能力。虽然传统的RAG系统通常使用向量数据库进行相似性搜索，但将图数据库集成到RAG流程中，可以带来显著的优势，尤其是在处理复杂关系和需要多跳推理的场景中。

为什么使用图数据库？

图数据库以节点和边的形式存储数据，这使得它们非常适合表示实体之间的关系。与向量数据库不同，图数据库能够执行精确的、结构化的查询，并且可以轻松地遍历关系。这种能力使得图数据库在以下方面表现出色：

RAG与图数据库的集成

将图数据库集成到RAG系统中，通常涉及以下步骤：

  1. 构建知识图谱：从结构化或非结构化数据中提取实体和关系，构建图。
  2. 创建向量索引：为图中的节点或关系生成嵌入向量，并存储在图数据库中或关联的向量索引中。
  3. 检索阶段：根据用户查询，使用向量相似性搜索找到相关节点，然后通过图遍历获取上下文。
  4. 生成阶段：将检索到的上下文与原始查询一起提供给LLM，生成最终答案。

示例：使用Neo4j实现RAG

让我们通过一个简单的示例来演示如何使用Neo4j（一种流行的图数据库）实现RAG。

步骤1：构建知识图谱

假设我们有一个关于电影和演员的数据集。我们可以创建节点来表示电影和演员，并创建边来表示“出演”关系。

CREATE (m:Movie {title: 'Inception', year: 2010})
CREATE (a:Actor {name: 'Leonardo DiCaprio'})
CREATE (a)-[:ACTED_IN]->(m)

步骤2：生成嵌入并存储

对于每个节点，我们可以使用预训练模型（如OpenAI的嵌入模型）生成向量，并将其存储为节点属性。

from openai import OpenAI
client = OpenAI()

def get_embedding(text):
    response = client.embeddings.create(model="text-embedding-ada-002", input=text)
    return response.data[0].embedding

# 为电影节点添加嵌入
movie = graph.run("MATCH (m:Movie {title: 'Inception'}) RETURN m").data()[0]['m']
embedding = get_embedding(movie['title'])
graph.run("MATCH (m:Movie {title: 'Inception'}) SET m.embedding = $emb", emb=embedding)

步骤3：检索相关节点

当用户提出查询时，我们首先使用向量相似性搜索找到最相关的节点。在Neo4j中，我们可以使用gds.similarity.cosine函数或通过外部向量索引（如Elasticsearch）来实现。

query = "Who starred in Inception?"
query_embedding = get_embedding(query)
# 使用向量索引查找相似节点
results = graph.run("""
CALL db.index.vector.queryNodes('movie_embeddings', 10, $query_embedding)
YIELD node, score
RETURN node.title, score
""", query_embedding=query_embedding)

步骤4：图遍历获取上下文

一旦我们找到了相关节点（例如“Inception”），我们可以通过图遍历获取其邻居节点，以提供更丰富的上下文。

MATCH (m:Movie {title: 'Inception'})-[:ACTED_IN]-(actor)
RETURN actor.name

步骤5：生成答案

最后，我们将检索到的上下文（例如演员列表）与原始查询一起发送给LLM，生成最终答案。

context = "Actors in Inception: Leonardo DiCaprio, ..."
prompt = f"Question: {query}\nContext: {context}\nAnswer:"
response = client.completions.create(model="gpt-4", prompt=prompt)
print(response.choices[0].text)

实际应用与挑战

将图数据库与RAG结合，在以下场景中特别有用：

然而，这种方法也带来一些挑战：

结论

将图数据库集成到RAG系统中，可以显著增强模型处理复杂关系的能力，并提供更好的可解释性。通过结合向量搜索和图遍历，我们可以构建更强大、更可靠的问答系统。随着图数据库和LLM技术的不断发展，这种集成将成为构建下一代AI应用的关键技术。

本笔记本展示了如何将LLM与Neo4j（一种图数据库）结合使用，以执行检索增强生成（RAG）。

为什么使用RAG？

如果您希望使用LLM基于您自己的内容或知识库生成答案，而不是在提示模型时提供大量上下文，您可以在数据库中获取相关信息，并使用这些信息来生成响应。

这使您能够：

为什么使用图数据库？

如果您的数据中数据点之间的关系很重要，并且您可能希望利用这些关系，那么考虑使用图数据库而不是传统的关系数据库可能是值得的。

图数据库擅长解决以下问题：

使用场景

图数据库特别适用于推荐系统、网络关系或分析数据点之间的相关性。

使用图数据库进行RAG的示例用例包括：

根据您的用例，您可以评估使用图数据库是否有意义。

在本笔记本中，我们将构建一个产品推荐聊天机器人，使用包含亚马逊产品数据的图数据库。

设置

我们首先安装并导入相关库。

确保你已经设置好 OpenAI 账户，并准备好你的 OpenAI API 密钥。

# Optional: run to install the libraries locally if you haven't already
!pip3 install langchain
!pip3 install openai
!pip3 install neo4j

import os
import json
import pandas as pd

# Optional: run to load environment variables from a .env file.
# This is not required if you have exported your env variables in another way or if you set it manually
!pip3 install python-dotenv
from dotenv import load_dotenv
load_dotenv()

# Set the OpenAI API key env variable manually
# os.environ["OPENAI_API_KEY"] = "<your_api_key>"

# print(os.environ["OPENAI_API_KEY"])

数据集

我们将使用一个从关系数据库创建并转换为 JSON 格式的数据集，通过 completions API 在实体之间建立关系。

然后，我们将把这些数据加载到图数据库中，以便能够对其进行查询。

加载数据集

# Loading a json dataset from a file
file_path = 'data/amazon_product_kg.json'

with open(file_path, 'r') as file:
    jsonData = json.load(file)

df =  pd.read_json(file_path)
df.head()

连接到数据库

# DB credentials
url = "bolt://localhost:7687"
username ="neo4j"
password = "<your_password_here>"

from langchain.graphs import Neo4jGraph

graph = Neo4jGraph(
    url=url, 
    username=username, 
    password=password
)

导入数据

def sanitize(text):
    text = str(text).replace("'","").replace('"','').replace('{','').replace('}', '')
    return text

# Loop through each JSON object and add them to the db
i = 1
for obj in jsonData:
    print(f"{i}. {obj['product_id']} -{obj['relationship']}-> {obj['entity_value']}")
    i+=1
    query = f'''
        MERGE (product:Product {{id: {obj['product_id']}}})
        ON CREATE SET product.name = "{sanitize(obj['product'])}", 
                       product.title = "{sanitize(obj['TITLE'])}", 
                       product.bullet_points = "{sanitize(obj['BULLET_POINTS'])}", 
                       product.size = {sanitize(obj['PRODUCT_LENGTH'])}

        MERGE (entity:{obj['entity_type']} {{value: "{sanitize(obj['entity_value'])}"}})

        MERGE (product)-[:{obj['relationship']}]->(entity)
        '''
    graph.query(query)

查询数据库

创建向量索引

为了高效地在数据库中搜索与用户查询密切相关的术语，我们需要使用嵌入。为此，我们将在每种类型的属性上创建向量索引。

我们将使用 Langchain 的 OpenAIEmbeddings 工具。需要注意的是，Langchain 添加了一个预处理步骤，因此生成的嵌入将与直接使用 OpenAI 嵌入 API 生成的嵌入略有不同。

from langchain.vectorstores.neo4j_vector import Neo4jVector
from langchain.embeddings.openai import OpenAIEmbeddings
embeddings_model = "text-embedding-3-small"

vector_index = Neo4jVector.from_existing_graph(
    OpenAIEmbeddings(model=embeddings_model),
    url=url,
    username=username,
    password=password,
    index_name='products',
    node_label="Product",
    text_node_properties=['name', 'title'],
    embedding_node_property='embedding',
)

def embed_entities(entity_type):
    vector_index = Neo4jVector.from_existing_graph(
        OpenAIEmbeddings(model=embeddings_model),
        url=url,
        username=username,
        password=password,
        index_name=entity_type,
        node_label=entity_type,
        text_node_properties=['value'],
        embedding_node_property='embedding',
    )
    
entities_list = df['entity_type'].unique()

for t in entities_list:
    embed_entities(t)

直接查询数据库

使用 GraphCypherQAChain，我们可以通过自然语言对数据库生成查询。

from langchain.chains import GraphCypherQAChain
from langchain.chat_models import ChatOpenAI

chain = GraphCypherQAChain.from_llm(
    ChatOpenAI(temperature=0), graph=graph, verbose=True,
)

chain.run("""
Help me find curtains
""")

从提示中提取实体

然而，与直接编写Cypher查询相比，这里几乎没有增加价值，而且容易出错。

实际上，让LLM直接生成Cypher查询可能会导致使用错误的参数，无论是实体类型还是关系类型，如上例所示。

我们将转而使用LLM来决定搜索什么，然后使用模板生成相应的Cypher查询。

为此，我们将指示我们的模型在用户提示中找到可用于查询数据库的相关实体。

entity_types = {
    "product": "Item detailed type, for example 'high waist pants', 'outdoor plant pot', 'chef kitchen knife'",
    "category": "Item category, for example 'home decoration', 'women clothing', 'office supply'",
    "characteristic": "if present, item characteristics, for example 'waterproof', 'adhesive', 'easy to use'",
    "measurement": "if present, dimensions of the item", 
    "brand": "if present, brand of the item",
    "color": "if present, color of the item",
    "age_group": "target age group for the product, one of 'babies', 'children', 'teenagers', 'adults'. If suitable for multiple age groups, pick the oldest (latter in the list)."
}

relation_types = {
    "hasCategory": "item is of this category",
    "hasCharacteristic": "item has this characteristic",
    "hasMeasurement": "item is of this measurement",
    "hasBrand": "item is of this brand",
    "hasColor": "item is of this color", 
    "isFor": "item is for this age_group"
 }

entity_relationship_match = {
    "category": "hasCategory",
    "characteristic": "hasCharacteristic",
    "measurement": "hasMeasurement", 
    "brand": "hasBrand",
    "color": "hasColor",
    "age_group": "isFor"
}

system_prompt = f'''
    You are a helpful agent designed to fetch information from a graph database. 
    
    The graph database links products to the following entity types:
    {json.dumps(entity_types)}
    
    Each link has one of the following relationships:
    {json.dumps(relation_types)}

    Depending on the user prompt, determine if it possible to answer with the graph database.
        
    The graph database can match products with multiple relationships to several entities.
    
    Example user input:
    "Which blue clothing items are suitable for adults?"
    
    There are three relationships to analyse:
    1. The mention of the blue color means we will search for a color similar to "blue"
    2. The mention of the clothing items means we will search for a category similar to "clothing"
    3. The mention of adults means we will search for an age_group similar to "adults"
    
    
    Return a json object following the following rules:
    For each relationship to analyse, add a key value pair with the key being an exact match for one of the entity types provided, and the value being the value relevant to the user query.
    
    For the example provided, the expected output would be:
    {{
        "color": "blue",
        "category": "clothing",
        "age_group": "adults"
    }}
    
    If there are no relevant entities in the user prompt, return an empty json object.
'''

print(system_prompt)

from openai import OpenAI
client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY", "<your OpenAI API key if not set as env var>"))

# Define the entities to look for
def define_query(prompt, model="gpt-4o"):
    completion = client.chat.completions.create(
        model=model,
        temperature=0,
        response_format= {
            "type": "json_object"
        },
    messages=[
        {
            "role": "system",
            "content": system_prompt
        },
        {
            "role": "user",
            "content": prompt
        }
        ]
    )
    return completion.choices[0].message.content

example_queries = [
    "Which pink items are suitable for children?",
    "Help me find gardening gear that is waterproof",
    "I'm looking for a bench with dimensions 100x50 for my living room"
]

for q in example_queries:
    print(f"Q: '{q}'\n{define_query(q)}\n")

生成查询

既然我们知道了要查找什么，就可以生成相应的 Cypher 查询来查询我们的数据库。

然而，提取出的实体可能与我们的数据不完全匹配，因此我们将使用 GDS 余弦相似度函数来返回与用户查询中实体具有相似关系的产品。

def create_embedding(text):
    result = client.embeddings.create(model=embeddings_model, input=text)
    return result.data[0].embedding

# The threshold defines how closely related words should be. Adjust the threshold to return more or less results
def create_query(text, threshold=0.81):
    query_data = json.loads(text)
    # Creating embeddings
    embeddings_data = []
    for key, val in query_data.items():
        if key != 'product':
            embeddings_data.append(f"${key}Embedding AS {key}Embedding")
    query = "WITH " + ",\n".join(e for e in embeddings_data)
    # Matching products to each entity
    query += "\nMATCH (p:Product)\nMATCH "
    match_data = []
    for key, val in query_data.items():
        if key != 'product':
            relationship = entity_relationship_match[key]
            match_data.append(f"(p)-[:{relationship}]->({key}Var:{key})")
    query += ",\n".join(e for e in match_data)
    similarity_data = []
    for key, val in query_data.items():
        if key != 'product':
            similarity_data.append(f"gds.similarity.cosine({key}Var.embedding, ${key}Embedding) > {threshold}")
    query += "\nWHERE "
    query += " AND ".join(e for e in similarity_data)
    query += "\nRETURN p"
    return query

def query_graph(response):
    embeddingsParams = {}
    query = create_query(response)
    query_data = json.loads(response)
    for key, val in query_data.items():
        embeddingsParams[f"{key}Embedding"] = create_embedding(val)
    result = graph.query(query, params=embeddingsParams)
    return result

example_response = '''{
    "category": "clothes",
    "color": "blue",
    "age_group": "adults"
}'''

result = query_graph(example_response)

# Result
print(f"Found {len(result)} matching product(s):\n")
for r in result:
    print(f"{r['p']['name']} ({r['p']['id']})")

查找相似物品

然后，我们可以利用图数据库根据共同特征查找相似的产品。

这正是图数据库真正发挥作用的地方。

例如，我们可以查找属于同一类别且具有另一个共同特征的产品，或者查找与相同实体有关联的产品。

这个标准是任意的，完全取决于在你的用例中什么最相关。

# Adjust the relationships_threshold to return products that have more or less relationships in common
def query_similar_items(product_id, relationships_threshold = 3):
    
    similar_items = []
        
    # Fetching items in the same category with at least 1 other entity in common
    query_category = '''
            MATCH (p:Product {id: $product_id})-[:hasCategory]->(c:category)
            MATCH (p)-->(entity)
            WHERE NOT entity:category
            MATCH (n:Product)-[:hasCategory]->(c)
            MATCH (n)-->(commonEntity)
            WHERE commonEntity = entity AND p.id <> n.id
            RETURN DISTINCT n;
        '''
    

    result_category = graph.query(query_category, params={"product_id": int(product_id)})
    #print(f"{len(result_category)} similar items of the same category were found.")
          
    # Fetching items with at least n (= relationships_threshold) entities in common
    query_common_entities = '''
        MATCH (p:Product {id: $product_id})-->(entity),
            (n:Product)-->(entity)
            WHERE p.id <> n.id
            WITH n, COUNT(DISTINCT entity) AS commonEntities
            WHERE commonEntities >= $threshold
            RETURN n;
        '''
    result_common_entities = graph.query(query_common_entities, params={"product_id": int(product_id), "threshold": relationships_threshold})
    #print(f"{len(result_common_entities)} items with at least {relationships_threshold} things in common were found.")

    for i in result_category:
        similar_items.append({
            "id": i['n']['id'],
            "name": i['n']['name']
        })
            
    for i in result_common_entities:
        result_id = i['n']['id']
        if not any(item['id'] == result_id for item in similar_items):
            similar_items.append({
                "id": result_id,
                "name": i['n']['name']
            })
    return similar_items

product_ids = ['1519827', '2763742']

for product_id in product_ids:
    print(f"Similar items for product #{product_id}:\n")
    result = query_similar_items(product_id)
    print("\n")
    for r in result:
        print(f"{r['name']} ({r['id']})")
    print("\n\n")

最终结果

现在所有部分都已就绪，我们将把它们整合在一起。

我们还可以添加一个回退选项，当在用户提示中找不到相关实体时，执行产品名称/标题相似度搜索。

我们将探讨两种方案：一种是使用Langchain代理实现对话式体验，另一种是仅基于代码的确定性方案。

根据您的使用场景，您可以选择其中一种方案，并根据需要进行调整。

def query_db(params):
    matches = []
    # Querying the db
    result = query_graph(params)
    for r in result:
        product_id = r['p']['id']
        matches.append({
            "id": product_id,
            "name":r['p']['name']
        })
    return matches    

def similarity_search(prompt, threshold=0.8):
    matches = []
    embedding = create_embedding(prompt)
    query = '''
            WITH $embedding AS inputEmbedding
            MATCH (p:Product)
            WHERE gds.similarity.cosine(inputEmbedding, p.embedding) > $threshold
            RETURN p
            '''
    result = graph.query(query, params={'embedding': embedding, 'threshold': threshold})
    for r in result:
        product_id = r['p']['id']
        matches.append({
            "id": product_id,
            "name":r['p']['name']
        })
    return matches

prompt_similarity = "I'm looking for nice curtains"
print(similarity_search(prompt_similarity))

构建 Langchain 智能体

我们将创建一个 Langchain 智能体来处理对话，并向用户询问更多上下文。

我们需要精确定义智能体的行为方式，并让它能够访问我们的查询和相似性搜索工具。

from langchain.agents import Tool, AgentExecutor, LLMSingleActionAgent, AgentOutputParser
from langchain.schema import AgentAction, AgentFinish, HumanMessage, SystemMessage


tools = [
    Tool(
        name="Query",
        func=query_db,
        description="Use this tool to find entities in the user prompt that can be used to generate queries"
    ),
    Tool(
        name="Similarity Search",
        func=similarity_search,
        description="Use this tool to perform a similarity search with the products in the database"
    )
]

tool_names = [f"{tool.name}: {tool.description}" for tool in tools]

from langchain.prompts import StringPromptTemplate
from typing import Callable


prompt_template = '''Your goal is to find a product in the database that best matches the user prompt.
You have access to these tools:

{tools}

Use the following format:

Question: the input prompt from the user
Thought: you should always think about what to do
Action: the action to take (refer to the rules below)
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

Rules to follow:

1. Start by using the Query tool with the prompt as parameter. If you found results, stop here.
2. If the result is an empty array, use the similarity search tool with the full initial user prompt. If you found results, stop here.
3. If you cannot still cannot find the answer with this, probe the user to provide more context on the type of product they are looking for. 

Keep in mind that we can use entities of the following types to search for products:

{entity_types}.

3. Repeat Step 1 and 2. If you found results, stop here.

4. If you cannot find the final answer, say that you cannot help with the question.

Never return results if you did not find any results in the array returned by the query tool or the similarity search tool.

If you didn't find any result, reply: "Sorry, I didn't find any suitable products."

If you found results from the database, this is your final answer, reply to the user by announcing the number of results and returning results in this format (each new result should be on a new line):

name_of_the_product (id_of_the_product)"

Only use exact names and ids of the products returned as results when providing your final answer.


User prompt:
{input}

{agent_scratchpad}

'''

# Set up a prompt template
class CustomPromptTemplate(StringPromptTemplate):
    # The template to use
    template: str
        
    def format(self, **kwargs) -> str:
        # Get the intermediate steps (AgentAction, Observation tuples)
        # Format them in a particular way
        intermediate_steps = kwargs.pop("intermediate_steps")
        thoughts = ""
        for action, observation in intermediate_steps:
            thoughts += action.log
            thoughts += f"\nObservation: {observation}\nThought: "
        # Set the agent_scratchpad variable to that value
        kwargs["agent_scratchpad"] = thoughts
        ############## NEW ######################
        #tools = self.tools_getter(kwargs["input"])
        # Create a tools variable from the list of tools provided
        kwargs["tools"] = "\n".join(
            [f"{tool.name}: {tool.description}" for tool in tools]
        )
        # Create a list of tool names for the tools provided
        kwargs["tool_names"] = ", ".join([tool.name for tool in tools])
        kwargs["entity_types"] = json.dumps(entity_types)
        return self.template.format(**kwargs)


prompt = CustomPromptTemplate(
    template=prompt_template,
    tools=tools,
    input_variables=["input", "intermediate_steps"],
)

from typing import List, Union
import re

class CustomOutputParser(AgentOutputParser):
    
    def parse(self, llm_output: str) -> Union[AgentAction, AgentFinish]:
        
        # Check if agent should finish
        if "Final Answer:" in llm_output:
            return AgentFinish(
                # Return values is generally always a dictionary with a single `output` key
                # It is not recommended to try anything else at the moment :)
                return_values={"output": llm_output.split("Final Answer:")[-1].strip()},
                log=llm_output,
            )
        
        # Parse out the action and action input
        regex = r"Action: (.*?)[\n]*Action Input:[\s]*(.*)"
        match = re.search(regex, llm_output, re.DOTALL)
        
        # If it can't parse the output it raises an error
        # You can add your own logic here to handle errors in a different way i.e. pass to a human, give a canned response
        if not match:
            raise ValueError(f"Could not parse LLM output: `{llm_output}`")
        action = match.group(1).strip()
        action_input = match.group(2)
        
        # Return the action and action input
        return AgentAction(tool=action, tool_input=action_input.strip(" ").strip('"'), log=llm_output)
    
output_parser = CustomOutputParser()

from langchain.chat_models import ChatOpenAI
from langchain import LLMChain
from langchain.agents.output_parsers.openai_tools import OpenAIToolsAgentOutputParser


llm = ChatOpenAI(temperature=0, model="gpt-4o")

# LLM chain consisting of the LLM and a prompt
llm_chain = LLMChain(llm=llm, prompt=prompt)

# Using tools, the LLM chain and output_parser to make an agent
tool_names = [tool.name for tool in tools]

agent = LLMSingleActionAgent(
    llm_chain=llm_chain, 
    output_parser=output_parser,
    stop=["\Observation:"], 
    allowed_tools=tool_names
)


agent_executor = AgentExecutor.from_agent_and_tools(agent=agent, tools=tools, verbose=True)

def agent_interaction(user_prompt):
    agent_executor.run(user_prompt)

prompt1 = "I'm searching for pink shirts"
agent_interaction(prompt1)

prompt2 = "Can you help me find a toys for my niece, she's 8"
agent_interaction(prompt2)

prompt3 = "I'm looking for nice curtains"
agent_interaction(prompt3)

构建纯代码体验

正如我们的实验所示，对于这类任务，使用代理可能不是最佳选择。

实际上，代理似乎从工具中检索结果，但会编造出虚假的响应。

对于这个特定用例，如果对话方面不太重要，我们实际上可以创建一个函数，该函数将调用我们先前定义的任务并提供答案。

import logging

def answer(prompt, similar_items_limit=10):
    print(f'Prompt: "{prompt}"\n')
    params = define_query(prompt)
    print(params)
    result = query_db(params)
    print(f"Found {len(result)} matches with Query function.\n")
    if len(result) == 0:
        result = similarity_search(prompt)
        print(f"Found {len(result)} matches with Similarity search function.\n")
        if len(result) == 0:
            return "I'm sorry, I did not find a match. Please try again with a little bit more details."
    print(f"I have found {len(result)} matching items:\n")
    similar_items = []
    for r in result:
        similar_items.extend(query_similar_items(r['id']))
        print(f"{r['name']} ({r['id']})")
    print("\n")
    if len(similar_items) > 0:
        print("Similar items that might interest you:\n")
        for i in similar_items[:similar_items_limit]:
            print(f"{i['name']} ({i['id']})")
    print("\n\n\n")
    return result

prompt1 = "I'm looking for food items to gift to someone for Christmas. Ideally chocolate."
answer(prompt1)

prompt2 = "Help me find women clothes for my wife. She likes blue."
answer(prompt2)

prompt3 = "I'm looking for nice things to decorate my living room."
answer(prompt3)

prompt4 = "Can you help me find a gift for my niece? She's 8 and she likes pink."
answer(prompt4)

结论

用户体验

当主要目标是从数据库中提取特定信息时，大型语言模型（LLM）可以显著增强我们的查询能力。

然而，为了确保用户体验万无一失，将大部分过程建立在稳健的代码逻辑上至关重要。

要打造一个真正对话式的聊天机器人，需要在提示工程方面进行进一步探索，可能包括少量示例。这种方法有助于降低生成不准确或误导性信息的风险，并确保更精确的响应。

最终，设计选择取决于期望的用户体验。例如，如果目标是创建视觉推荐系统，那么对话界面的重要性就相对较低。

使用知识图谱

从知识图谱中检索内容增加了复杂性，但如果您想利用项目之间的连接，这会很有用。

本笔记本的查询部分同样适用于关系数据库，而当我们希望将结果与图谱中浮现的相似项目耦合时，知识图谱就派上了用场。

考虑到增加的复杂性，请确保使用知识图谱是您用例的最佳选择。

如果是这样，请随意优化本食谱所呈现的内容，以匹配您的需求并实现更好的性能！

常见问题

常见错误：1）实体提取不准确，导致查询无结果——可调整系统提示词，增加示例，或使用few-shot学习；2）Cypher查询语法错误——建议使用参数化查询，避免字符串拼接；3）向量索引未正确创建——检查节点标签和属性名是否匹配；4）相似度阈值过高导致结果为空——适当降低阈值（如0.8）；5）Langchain Agent产生虚假响应——改用纯代码方案或增加验证步骤。排查方法：使用Neo4j Browser手动测试Cypher查询；打印中间结果（如实体提取输出、查询语句）；监控API调用日志。生产环境优化：使用批量嵌入和索引构建；对图查询进行性能调优（如添加索引）；采用缓存机制减少重复计算；考虑使用图数据库的并行查询能力。

企业应用扩展

可扩展方向：1）支持更多实体类型和关系，如产品评论、用户画像；2）引入动态图更新机制，实时反映数据变化；3）结合知识图谱嵌入（如TransE）增强关系推理；4）使用更先进的混合检索策略，如RAG-Fusion；5）针对特定领域（如医疗、金融）定制实体提取模型；6）优化LLM提示，实现多轮对话中的上下文记忆；7）部署为微服务，提供REST API接口。

总结

本文深入探讨了如何将图数据库（以Neo4j为例）集成到RAG系统中，以增强大语言模型处理复杂关系和多跳推理的能力。文章从图数据库的优势出发，详细介绍了构建知识图谱、生成向量索引、混合检索（向量相似性+图遍历）以及生成答案的完整流程。通过一个产品推荐聊天机器人的实战案例，展示了使用Langchain和OpenAI API实现实体提取、Cypher查询生成、相似物品推荐等关键步骤。同时，文章对比了基于Langchain Agent的对话式方案与纯代码确定性方案的优劣，并提供了常见错误排查与生产环境优化建议，帮助开发者构建更可靠、可解释的AI应用。

内容来源与验证

继续系统学习

站内还有更多完整实战教程

加入会员，解锁全部教程、完整代码与持续更新内容。

加入会员
查看全部RAG 与知识库教程