85.图RAG-llamaindex图谱关系提取器-Schema架构LLM路径提取器

内容参考于图灵AI大模型全栈图谱关系提取器提取器类型用途实现逻辑是否调用LLM推荐指数SimpleLLMPathExtractor三元组抽取快速构建知识图谱Prompt → LLM → 抽取 Entity-Relation-Entity✅⭐⭐DynamicLLMPathExtractor动态图谱抽取自动发现实体类型和关系类型Prompt → LLM → 实体分类 → 关系推断✅⭐⭐⭐SchemaLLMPathExtractorSchema约束抽取企业级GraphRAGPrompt Schema → LLM → 验证 → 输出✅⭐⭐⭐⭐⭐ImplicitPathExtractor隐式关系抽取保留文档结构关系根据Node Relationship自动生成边❌⭐⭐⭐⭐SchemaLLMPathExtractor它可以让LLM大模型严格按照我们预定义的实体类型Entities、关系类型Relations、它们之间的连接规则来提取数据之前的提取器都是按着大模型的逻辑来提取DynamicLLMPathExtractor提取器也只是让我们控制一部分这个可以让我们之间控制大模型的逻辑必须按着我们的逻辑来提取但是SchemaLLMPathExtractor提取器就需要我们自己考虑很多东西如果不需要非常严谨使用DynamicLLMPathExtractor提取器比较好它的逻辑首先它还是先对文档进行分割然后我们定义结构Schema也就是定义实体类型、关系类型、路径约束实体类型比如必须有人物、公司、产品等关系类型必须有创立、就职于、拥有等路径约束比如什么人物创建了什么公司、什么人物就职于什么公司、什么人物拥有什么公司这样的内容然后它把我们提供的 实体类型、关系类型、路径约束 内容拼接成提示词比如 允许的实体类型人物、公司、产品允许的关系类型创立、就职于、拥有有效的三元组模式 什么人物创建了什么公司、什么人物就职于什么公司、什么人物拥有什么公司提取完后它会通过Pydantic进行校验校验当前的数据是否跟我们的逻辑一致我们自己设定的是否严格模式来决定一致就存储不一致就丢掉然后构建知识图谱然后存储图数据它的缺点就是必须对业务和文档很熟悉才可以写好实体类型、关系类型、路径约束如果写不好就利用大模型写把核心的文档内容给大模型让大模型进行提取 实体类型、关系类型、路径约束 的限制SchemaLLMPathExtractor使用的提示词按着CTRL鼠标左键单击下图红框按着CTRL鼠标左键单击下图红框下图红框就是它使用的提示词中文翻译可以看出 它的提示词很简单提示词简单了生成的内容就不好所以真正用的时候需要我们自己重写提示词如果我们不传递 实体类型、关系类型、路径约束 如下图红框它也有默认的对应的内容如下图红框如下图红框它这样生成的还是有点不好这就说明我们写的 实体类型、关系类型、路径约束 不够好默认的提示词也不够好下图是自定义提示词抽取的内容自定义提示词也是让大模型生成的如下图代码运行效果图代码import asyncio from pathlib import Path from typing import Literal from llama_index.core import ( SimpleDirectoryReader, Settings, ) from llama_index.core.node_parser import SentenceSplitter from llama_index.core.indices.property_graph import ( PropertyGraphIndex, SchemaLLMPathExtractor, ) from llama_index.core.ingestion import ( IngestionPipeline, IngestionCache, ) from llama_index.core.storage.kvstore import SimpleKVStore from llama_index.core.storage.chat_store import SimpleChatStore from llama_index.core.memory import ChatMemoryBuffer from llama_index.graph_stores.neo4j import Neo4jPropertyGraphStore from base_llm import llm, embed_model # 全局配置 Settings.llm llm Settings.embed_model embed_model # 创建Neo4j数据库连接 graph_store Neo4jPropertyGraphStore( usernameneo4j, password11111111, urlbolt://localhost:7687, ) # 读取文档 documents SimpleDirectoryReader(input_files[./data_file/小说.txt]).load_data() # 摄取缓存 CACHE_PATH ./cache/cache.json cache_file Path(CACHE_PATH) # 创建SimpleKVStore缓存 if cache_file.is_file(): print(缓存文件存在) kvstore SimpleKVStore.from_persist_path(CACHE_PATH) else: print(缓存文件不存在) kvstore SimpleKVStore() cache IngestionCache(cachekvstore) # # 实体约束必须对业务很熟悉才可以写好 # entities Literal[ # # 人物 # PERSON, # # 家族 / 家庭 # FAMILY, # # 地点 # PLACE, # # 技能 # SKILL, # ] # # # 关系约束必须对业务很熟悉才可以写好 # relations Literal[ # # 父子关系 # FATHER_OF, # # 兄弟关系 # BROTHER_OF, # # 居住于 # LIVES_IN, # # 拥有技能 # HAS_SKILL, # ] # # # 路径约束必须对业务很熟悉才可以写好 # validation_schema [ # # 人物之间的父子关系 # (PERSON, FATHER_OF, PERSON), # # 人物之间的兄弟关系 # (PERSON, BROTHER_OF, PERSON), # # 人物居住的地点 # (PERSON, LIVES_IN, PLACE), # # 人物拥有的技能 # (PERSON, HAS_SKILL, SKILL), # ] # # # 创建图谱路径抽取器默认提示词 # kg_extractor SchemaLLMPathExtractor( # llmllm, # # 设置实体 # possible_entitiesentities, # # 设置关系 # possible_relationsrelations, # # 设置路径 # kg_validation_schemavalidation_schema, # # 设定的关系丢弃True会丢弃False不丢弃 # strictTrue, # # 每个切分的文本抽多少个关系10个是很少的如果是1024个字符的内容一般要抽取出 300到400左右关系三分之一 # max_triplets_per_chunk10, # # 并发数 # num_workers4, # ) # 2. 自定义 Schema保持简洁避免多余属性 MY_ENTITIES Literal[人物, 门派, 武功, 地点, 物品] MY_RELATIONS Literal[修炼, 隶属于, 创立, 位于, 拥有] MY_VALIDATION_SCHEMA [ (人物, 修炼, 武功), (人物, 隶属于, 门派), (门派, 位于, 地点), ] # 2. 自定义提示词中文 业务规则补充 CUSTOM_EXTRACT_PROMPT 你是专业的知识图谱抽取专家请从以下文本中抽取符合要求的三元组。 【抽取规则】 1. 实体类型仅限人物、门派、武功、地点、物品、事件 2. 关系类型仅限修炼、隶属于、创立、位于、拥有、参与 3. 仅抽取明确存在的事实禁止推断和脑补 4. 实体名称要完整、规范不要用代词他、她、它 5. 严格遵守合法的实体-关系组合 - 人物 可以 修炼/隶属于/参与/拥有 武功/门派/事件/物品 - 门派 可以 位于/创立 地点/武功 - 事件 可以 发生于 地点 - 物品 可以 位于 地点 请严格控制输出数量最多提取 {max_triplets_per_chunk} 条三元组。 ------- 待提取文本 {text} ------- # 3. 初始化 SchemaLLMPathExtractor kg_extractor SchemaLLMPathExtractor( llmllm, # 你的 LLM 实例 # 自定义提示词 extract_promptCUSTOM_EXTRACT_PROMPT, # 自定义实体枚举 possible_entitiesMY_ENTITIES, # 自定义关系枚举 possible_relationsMY_RELATIONS, # 自定义路径校验规则 kg_validation_schemaMY_VALIDATION_SCHEMA, # 开启严格校验必开否则 schema 不生效 strictTrue, # 其他参数按需设置 max_triplets_per_chunk10, num_workers4, ) # 创建摄取管道这里只进行分割这里会缓存分割后的文档 split_pipeline IngestionPipeline( transformations[ SentenceSplitter( chunk_size512, chunk_overlap50, ), ], cachecache, ) # 图谱抽取 Pipeline这里不缓存 EntityNode 和 Relation也就是不缓存实体和关系 kg_pipeline IngestionPipeline( transformations[ kg_extractor, ], ) # 执行摄取管道分割文档 split_nodes split_pipeline.run( documentsdocuments, show_progressTrue ) print(切分的文档数:, len(split_nodes)) # 保存切分缓存 kvstore.persist(CACHE_PATH) # 执行摄取管道生成知识图谱数据 nodes kg_pipeline.run( nodessplit_nodes, show_progressTrue ) print(节点数量:, len(nodes)) # 创建图索引 index PropertyGraphIndex( nodesnodes, property_graph_storegraph_store, llmllm, embed_modelembed_model, ) print(图谱构建完成) # 聊天记录存储 CHAT_STORE_PATH ./cache/chat_store.json if Path(CHAT_STORE_PATH).exists(): chat_store SimpleChatStore.from_persist_path(CHAT_STORE_PATH) print(聊天记录恢复成功) else: chat_store SimpleChatStore() print(创建新的聊天记录) # 聊天记忆 memory ChatMemoryBuffer.from_defaults( token_limit8000, chat_storechat_store, chat_store_keyuser_001 ) # 聊天引擎 chat_engine index.as_chat_engine( memorymemory, similarity_top_k5, verboseTrue, ) # 多轮聊天 while True: query input(\n用户) if query exit: break response chat_engine.chat(query) print(\nAI) print(response) # 持久化聊天记录 chat_store.persist(persist_pathCHAT_STORE_PATH) # 查看历史记录 history memory.get() print(\n聊天历史) for msg in history: print(msg.role) print(msg.content) print() # 关闭 Neo4j graph_store.close() asyncio.run(graph_store._async_driver.close())