低代码 RAG 只是信息搬运工，Graph RAG 让 AI 具备垂直深度推理能力！

# 安装必要库pip install neo4j langchain openai py2neo transformers spacypython -m spacy download en_core_web_sm

from py2neo import Graph, Node, Relationshipimport spacy
# 初始化 Neo4j 连接graph = Graph("bolt://localhost:7687", auth=("neo4j", "password"))
# 加载 spaCy 模型nlp = spacy.load("en_core_web_sm")
def extract_entities_relations(text):    """    从文本中提取实体和关系    """    doc = nlp(text)    entities = [ent.text for ent in doc.ents]  # 提取命名实体    relations = []
    # 简化的关系提取逻辑 - 实际应用中可使用更复杂的模型    for token in doc:        if token.dep_ in ("nsubj", "dobj"):  # 提取主语和宾语关系            relations.append((token.head.text, token.dep_, token.text))
    return entities, relations
def build_knowledge_graph(text_data):    """    构建知识图谱    """    for text in text_data:        entities, relations = extract_entities_relations(text)
        # 创建节点        nodes = {}        for entity in entities:            node = Node("Entity", name=entity)            graph.create(node)            nodes[entity] = node
        # 创建关系        for rel in relations:            head, relation, tail = rel            if head in nodes and tail in nodes:                relationship = Relationship(nodes[head], relation, nodes[tail])                graph.create(relationship)
# 示例数据texts = [    "智能体AI公众号主要写AI相关的文章",    "老谭是智能体AI公众号的作者",    "智能体AI公众号的作者一般在长沙"]
build_knowledge_graph(texts)

from langchain.graphs import Neo4jGraphfrom langchain.chains import GraphCypherQAChainfrom langchain.chat_models import ChatOpenAI
# 连接 Neo4jgraph = Neo4jGraph(    url="bolt://localhost:7687",    username="neo4j",    password="password")
# 定义检索函数def graph_retrieval(query):    """    使用 Cypher 查询从知识图谱中检索相关信息    """    # 自动生成 Cypher 查询    cypher_query = f"""    MATCH path = (e1)-[r]->(e2)    WHERE toLower(e1.name) CONTAINS toLower('{query}')        OR toLower(e2.name) CONTAINS toLower('{query}')       OR toLower(type(r)) CONTAINS toLower('{query}')    RETURN path    LIMIT 5    """
    # 执行查询    result = graph.query(cypher_query)
    # 格式化结果    context = []    for record in result:        path = record["path"]        nodes = path.nodes        rels = path.relationships
        for rel in rels:            start_node = nodes[rel.start]            end_node = nodes[rel.end]            context.append(                f"{start_node['name']} - {rel.type} - {end_node['name']}"            )
    return "\n".join(context) if context else "No relevant information found."

from langchain.prompts import PromptTemplatefrom langchain.chains import RetrievalQAfrom langchain.llms import OpenAI
# 定义提示模板GRAPH_PROMPT = PromptTemplate(    template="""基于以下知识图谱信息和问题，提供详细的回答。如果信息不足，请说明。
    知识图谱信息:    {context}
    问题: {question}
    回答:""",    input_variables=["context", "question"],)
# 创建 Graph RAG 链llm = OpenAI(temperature=0)graph_qa_chain = RetrievalQA.from_chain_type(    llm=llm,    chain_type="stuff",    retriever=graph_retrieval,  # 使用自定义的图检索函数    chain_type_kwargs={"prompt": GRAPH_PROMPT},    verbose=True)
# 使用示例question = "谁是智能体AI公众号的作者"result = graph_qa_chain.run(question)print(result)

from torch_geometric.data import Dataimport torchfrom transformers import AutoModel, AutoTokenizer
class GraphEmbedder:    def __init__(self):        self.tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")        self.model = AutoModel.from_pretrained("bert-base-uncased")
    def get_embeddings(self, text):        inputs = self.tokenizer(text, return_tensors="pt", padding=True, truncation=True)        with torch.no_grad():            outputs = self.model(**inputs)        return outputs.last_hidden_state.mean(dim=1)
    def create_graph_data(self, cypher_result):        """        将 Cypher 查询结果转换为 PyG 图数据        """        node_texts = []        edge_index = []        edge_attrs = []
        node_dict = {}        node_counter = 0
        for record in cypher_result:            path = record["path"]            nodes = path.nodes            rels = path.relationships
            for rel in rels:                start_node = nodes[rel.start]                end_node = nodes[rel.end]
                # 添加节点                if start_node["name"] not in node_dict:                    node_dict[start_node["name"]] = node_counter                    node_texts.append(start_node["name"])                    node_counter += 1
                if end_node["name"] not in node_dict:                    node_dict[end_node["name"]] = node_counter                    node_texts.append(end_node["name"])                    node_counter += 1
                # 添加边                edge_index.append([node_dict[start_node["name"]], node_dict[end_node["name"]]])                edge_attrs.append(rel.type)
        # 获取节点嵌入        node_embeddings = torch.cat([self.get_embeddings(text) for text in node_texts])
        # 创建图数据对象        data = Data(            x=node_embeddings,            edge_index=torch.tensor(edge_index, dtype=torch.long).t().contiguous(),            edge_attr=edge_attrs        )
        return data
# 使用示例embedder = GraphEmbedder()cypher_result = graph.query("MATCH path=(e1)-[r]->(e2) RETURN path LIMIT 10")graph_data = embedder.create_graph_data(cypher_result)

from fastapi import FastAPIfrom pydantic import BaseModel
app = FastAPI()
class Query(BaseModel):    question: str
@app.post("/ask")async def ask_question(query: Query):    # 1. 从知识图谱检索    context = graph_retrieval(query.question)
    # 2. 使用大模型生成回答    response = graph_qa_chain.run({        "question": query.question,        "context": context    })
    # 3. 返回结果    return {"answer": response}
# 运行命令: uvicorn app:app --reload