1. 项目概述从海量文档到实时洞察的挑战最近在做一个内部项目需要把公司过去几年积累的几百份技术文档、会议纪要和产品手册“盘活”。老板的要求很直接能不能做个系统让新来的同事问个问题比如“我们A产品的数据备份方案和B方案在成本上有啥区别”系统能立刻从这些文档里找到相关信息并且把产品、方案、成本这些概念之间的关系清晰地展示出来而不是扔给用户一堆需要自己再整理的搜索结果。这个需求本质上就是构建一个能够实时查询和推理的知识图谱系统。传统的知识图谱构建往往是个“离线批处理”的活儿。你需要用NLP工具从文档里抽取出实体比如产品名、技术名词和关系比如“包含”、“优于”、“依赖于”然后存进图数据库。这个过程耗时很长数据更新也不及时。而“实时”的要求意味着我们需要一种更敏捷的方式当用户提出一个新问题时系统能动态地从最新的文档中抽取知识并即时构建出一个针对该问题的、轻量级的图谱片段进行展示和推理。这就是我选择Graphiti这个框架进行实战探索的核心原因。Graphiti 并不是一个图数据库而是一个将大语言模型LLM的语义理解能力与图结构Graph的关联推理能力结合起来的开发框架。它的核心思路是“按需构图”。系统不会事先把所有可能的知识都抽取并存储成一张大图那成本高且维护难而是利用LLM理解用户查询的意图动态地决定需要从文档中抽取哪些实体和关系并即时组织成图谱进行回答。这对于处理海量、非结构化文档且需求多变的场景来说非常具有吸引力。接下来我将完整分享这次实战的笔记包括设计思路、关键实现、踩过的坑以及一些性能调优的心得。2. 核心架构与工具选型解析2.1 为什么是 Graphiti LLM 向量数据库的组合面对“海量文档实时知识图谱”的需求我评估了几个方案。传统方案是NLP流水线实体识别、关系抽取 图数据库Neo4j, NebulaGraph。这个方案的问题是流水线需要大量标注数据来训练或者依赖规则泛化能力差且构建的是“静态全图”任何文档更新都需要重新跑一遍流程无法实时。Graphiti提出的动态构图理念更符合我们的场景。其架构核心是LLM作为“图谱构建师”利用LLM强大的零样本/少样本理解能力将用户查询和文档片段转化为图谱查询指令或直接生成图谱结构。它替代了传统的训练好的NLP模型。向量数据库作为“记忆库”所有文档被切分成片段chunks编码成向量后存入向量数据库如Chroma, Weaviate。当用户提问时先将问题本身也转化为向量在向量库中进行相似性检索快速找到最相关的文档片段。这解决了“从海量文档中快速定位相关信息”的问题。Graphiti作为“协调中枢”Graphiti框架负责编排整个流程。它接收用户查询调用LLM分析查询意图并生成针对向量检索结果的“信息抽取指令”然后再调用LLM根据指令和检索到的文本抽取出实体和关系最后组织成图结构返回。我最终的技术栈如下框架GraphitiPythonLLM服务OpenAI GPT-4 API用于复杂意图理解和信息抽取。对于成本敏感的部分也用到了 GPT-3.5-Turbo。向量数据库ChromaDB轻量级易于集成适合原型和中小规模数据。开发语言Python 3.10。辅助工具LangChain用于文档加载、文本分割和部分链式调用编排但Graphiti本身也提供了类似的模式。注意LLM API的选择至关重要。GPT-4在理解复杂指令和进行精确抽取方面显著优于3.5但成本也高。我的策略是在构图的关键步骤如解析查询生成抽取指令使用GPT-4以保证质量在简单的文本摘要或初筛时使用GPT-3.5。2.2 环境准备与初始化首先建立一个干净的Python环境并安装核心依赖。# 创建并激活虚拟环境 python -m venv graphiti_env source graphiti_env/bin/activate # Linux/Mac # graphiti_env\Scripts\activate # Windows # 安装核心包 pip install graphiti-ai pip install openai pip install chromadb pip install langchain langchain-openai pip install pypdf # 用于读取PDF文档 pip install tiktoken # 用于计算Token控制成本接下来进行关键的初始化配置主要是设置LLM和向量数据库。import os from graphiti import Graphiti from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain_community.vectorstores import Chroma from langchain_text_splitters import RecursiveCharacterTextSplitter # 1. 设置OpenAI API密钥请替换为你的密钥或从环境变量读取 os.environ[OPENAI_API_KEY] your-api-key-here # 2. 初始化LLM客户端 # 用于对话和复杂推理的LLM llm_gpt4 ChatOpenAI(modelgpt-4, temperature0.1) # temperature调低使输出更确定 llm_gpt35 ChatOpenAI(modelgpt-3.5-turbo, temperature0.1) # 用于生成文本嵌入向量的模型 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 性价比高 # 3. 初始化Graphiti graphiti_client Graphiti() # 4. 初始化文本分割器 # 这里选择递归字符分割尝试保持段落和句子的完整性 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, # 每个片段约1000字符 chunk_overlap200, # 片段间重叠200字符避免信息被割裂 length_functionlen, separators[\n\n, \n, 。, , , , , , ] )实操心得chunk_size的设置是平衡检索精度和上下文完整性的关键。太小如200会导致信息碎片化LLM缺乏足够上下文进行准确抽取太大如2000则可能引入无关噪声降低向量检索的准确性。1000-1500是一个常见的起步值需要根据你的文档平均段落长度进行调整。chunk_overlap能有效缓解句子被腰斩的问题。3. 知识库构建文档处理与向量化知识图谱的“知识”来源于文档因此第一步是将非结构化的文档转化为结构化的、可检索的知识单元。3.1 文档加载与预处理我处理的文档包括PDF、Word和Markdown。使用LangChain的文档加载器可以统一处理。from langchain_community.document_loaders import PyPDFLoader, UnstructuredWordDocumentLoader, TextLoader from typing import List def load_documents(directory_path: str) - List[Document]: 加载指定目录下的所有支持格式的文档 documents [] for filename in os.listdir(directory_path): filepath os.path.join(directory_path, filename) if filename.endswith(.pdf): loader PyPDFLoader(filepath) elif filename.endswith(.docx): loader UnstructuredWordDocumentLoader(filepath) elif filename.endswith(.md) or filename.endswith(.txt): loader TextLoader(filepath, encodingutf-8) else: continue loaded_docs loader.load() # 为每个文档片段添加源文件元数据便于追溯 for doc in loaded_docs: doc.metadata[source] filename documents.extend(loaded_docs) return documents # 示例加载docs文件夹下的所有文档 raw_documents load_documents(./docs) print(f共加载了 {len(raw_documents)} 个原始文档片段。)3.2 文本分割与向量数据库持久化加载后的文档需要被分割成更小的片段然后转化为向量存入ChromaDB。def create_vector_store(documents: List[Document], persist_directory: str ./chroma_db): 将文档分割、向量化并持久化到ChromaDB。 # 1. 分割文本 print(开始分割文本...) all_splits text_splitter.split_documents(documents) print(f分割后得到 {len(all_splits)} 个文本片段。) # 2. 创建并持久化向量存储 print(开始生成向量并存入数据库...) vectorstore Chroma.from_documents( documentsall_splits, embeddingembeddings, persist_directorypersist_directory ) # 显式持久化 vectorstore.persist() print(f向量数据库已创建并保存至 {persist_directory}) return vectorstore # 执行创建 vector_store create_vector_store(raw_documents)注意事项向量数据库的持久化路径很重要。首次运行会创建后续运行可以直接加载无需重复向量化节省成本和时间。# 后续加载已有向量数据库 vector_store Chroma(persist_directory./chroma_db, embedding_functionembeddings)3.3 检索策略优化提升召回率简单的向量相似性检索有时会漏掉关键信息。我采用了混合检索策略来提升召回率。from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor def create_enhanced_retriever(vectorstore, llm_for_compressionllm_gpt35): 创建增强的检索器结合向量检索和上下文压缩。 # 基础向量检索器设置检索数量稍大 base_retriever vector_store.as_retriever(search_kwargs{k: 8}) # 使用LLM对检索结果进行压缩/重排只保留与查询最相关的部分 # 这可以节省后续LLM处理的Token并提升精度 compressor LLMChainExtractor.from_llm(llm_for_compression) compression_retriever ContextualCompressionRetriever( base_compressorcompressor, base_retrieverbase_retriever ) return compression_retriever enhanced_retriever create_enhanced_retriever(vector_store)这个LLMChainExtractor会在向量检索返回片段后再用一个小型LLM我用GPT-3.5快速扫描每个片段提取出其中与查询直接相关的句子过滤掉无关内容。实测下来这能有效提高最终构图时输入LLM的上下文质量。4. 动态图谱构建Graphiti 核心实战这是最核心的部分即如何利用Graphiti和LLM根据用户查询动态构建知识图谱。4.1 定义图谱模式Schema虽然我们是动态构图但提前定义一个期望的图谱模式能极大引导LLM的输出格式使其更结构化、更可控。我们定义实体类型和关系类型。# 根据我们的技术文档领域定义可能的实体和关系类型 GRAPH_SCHEMA { entity_types: [ 产品, 技术组件, 功能特性, 部署环境, 人员角色, 成本指标, 问题, 解决方案 ], relationship_types: [ 包含, 依赖于, 优于, 劣于, 导致, 解决, 拥有, 属于, 相关于, 成本为 ] } # 将这个模式转化为给LLM的提示词部分 schema_prompt f 你是一个知识图谱构建专家。请从给定的文本中提取信息构建一个知识图谱。 图谱中的节点实体类型应限于{, .join(GRAPH_SCHEMA[entity_types])}。 图谱中的边关系类型应限于{, .join(GRAPH_SCHEMA[relationship_types])}。 请以JSON格式输出包含entities和relationships两个列表。 每个实体应包含id唯一标识如‘产品_A’、name显示名称、type实体类型。 每个关系应包含source_id源实体ID、target_id目标实体ID、type关系类型、description可选关系描述。 4.2 实现动态构图管道现在我们将检索、LLM调用和结果解析串联起来。import json from langchain_core.prompts import ChatPromptTemplate def build_knowledge_graph(query: str, retriever, top_k: int 5) - dict: 核心函数根据用户查询动态构建知识图谱。 # 1. 检索相关文档片段 print(f检索与查询‘{query}’相关的文档...) relevant_docs retriever.invoke(query) context_text \n\n---\n\n.join([doc.page_content for doc in relevant_docs]) if not context_text: return {entities: [], relationships: [], context: 未找到相关信息。} # 2. 构建LLM提示词 prompt_template ChatPromptTemplate.from_messages([ (system, 你是一个精准的信息抽取助手。 schema_prompt), (human, 用户查询{query} 请基于以下相关文本内容抽取与查询意图紧密相关的实体和关系构建一个聚焦的知识图谱。 注意图谱应直接服务于回答该查询避免抽取无关的宽泛知识。 相关文本 {context} 请输出纯净的JSON。 ) ]) # 3. 调用LLM使用GPT-4以保证抽取质量 chain prompt_template | llm_gpt4 response chain.invoke({query: query, context: context_text}) # 4. 解析LLM的JSON输出 try: # LLM输出可能是带Markdown代码块的JSON需要清理 content response.content if json in content: content content.split(json)[1].split()[0].strip() elif in content: content content.split()[1].split()[0].strip() graph_data json.loads(content) # 添加检索到的源文档信息作为图谱元数据 graph_data[source_documents] [doc.metadata.get(source, unknown) for doc in relevant_docs] return graph_data except json.JSONDecodeError as e: print(fLLM返回的JSON解析失败: {e}) print(f原始返回内容: {response.content}) # 返回一个包含错误信息的空结构 return {entities: [], relationships: [], error: 图谱生成失败LLM返回格式异常。} # 示例查询 query_example “我们产品A的数据备份方案和产品B的容灾方案在成本上有何差异” result_graph build_knowledge_graph(query_example, enhanced_retriever) print(json.dumps(result_graph, indent2, ensure_asciiFalse))一个成功的输出可能如下所示{ entities: [ {id: 产品_A, name: 产品A, type: 产品}, {id: 备份方案_X, name: 基于快照的增量备份方案, type: 解决方案}, {id: 成本_月度1000, name: 每月1000元, type: 成本指标}, {id: 产品_B, name: 产品B, type: 产品}, {id: 容灾方案_Y, name: 跨地域热备容灾方案, type: 解决方案}, {id: 成本_月度5000, name: 每月5000元, type: 成本指标} ], relationships: [ {source_id: 产品_A, target_id: 备份方案_X, type: 拥有, description: 采用}, {source_id: 备份方案_X, target_id: 成本_月度1000, type: 成本为, description: null}, {source_id: 产品_B, target_id: 容灾方案_Y, type: 拥有, description: 采用}, {source_id: 容灾方案_Y, target_id: 成本_月度5000, type: 成本为, description: null}, {source_id: 备份方案_X, target_id: 容灾方案_Y, type: 劣于, description: 在容灾级别和成本上} ], source_documents: [产品A白皮书.pdf, 产品B技术架构.docx, 成本核算指南.md] }4.3 图谱可视化与交互生成JSON数据后我们可以用网络图库进行可视化。这里使用networkx和pyvis。import networkx as nx from pyvis.network import Network def visualize_graph(graph_data: dict, output_html_path: str knowledge_graph.html): 将图谱数据可视化为交互式HTML网页。 G nx.DiGraph() # 创建有向图 # 添加节点 for entity in graph_data.get(entities, []): G.add_node( entity[id], labelentity[name], titlef类型: {entity[type]}, groupentity[type] # 按类型分组便于可视化区分 ) # 添加边 for rel in graph_data.get(relationships, []): G.add_edge( rel[source_id], rel[target_id], titlerel[type] (f: {rel[description]} if rel.get(description) else ), labelrel[type] ) # 使用Pyvis生成交互式网络 net Network(height750px, width100%, directedTrue, notebookFalse) net.from_nx(G) # 可以调整一些物理布局参数让图更美观 net.set_options( var options { physics: { forceAtlas2Based: { gravitationalConstant: -50, centralGravity: 0.01, springLength: 100, springConstant: 0.08 }, minVelocity: 0.75, solver: forceAtlas2Based } } ) net.save_graph(output_html_path) print(f知识图谱已可视化保存至: {output_html_path}) # 在Jupyter Notebook中可以直接显示 # return net.show(f{output_html_path}) # 可视化上面生成的图谱 visualize_graph(result_graph)生成的HTML文件可以在浏览器中打开你可以拖动节点放大缩小清晰地看到“产品A-拥有-备份方案X-成本为-月度1000元”以及“劣于”关系连接到产品B的容灾方案。这种可视化对于呈现复杂关系非常直观。5. 高级技巧与性能优化5.1 缓存与异步处理频繁调用LLM API成本高、速度慢。对于相对稳定的文档库可以对“查询-检索结果-构图”进行缓存。import hashlib import pickle from functools import lru_cache def get_query_hash(query: str, top_k: int) - str: 生成查询的哈希键用于缓存 return hashlib.md5(f{query}_{top_k}.encode()).hexdigest() lru_cache(maxsize100) def cached_build_graph(query_hash: str, context_text: str) - dict: 缓存构图结果。注意context_text是检索到的文本如果文档库更新缓存需要失效。 这里为简化假设文档库短期内不变。生产环境需用Redis等外部缓存并设置过期。 # 这里模拟一个缓存查找实际应连接缓存数据库 cache_file f./cache/{query_hash}.pkl if os.path.exists(cache_file): with open(cache_file, rb) as f: return pickle.load(f) # 如果没有缓存则调用真正的构图函数这里需要原函数支持略作修改 # 实际应用中应将构图逻辑封装此处仅为示意 return None # 修改后的构图函数加入缓存逻辑 def build_knowledge_graph_with_cache(query: str, retriever, top_k: int 5, use_cacheTrue) - dict: query_hash get_query_hash(query, top_k) if use_cache: cached_result cached_build_graph(query_hash, ) # 需要更精细的缓存键设计 if cached_result: print(命中缓存) return cached_result # ... (原有的检索和构图逻辑) ... result build_knowledge_graph(query, retriever, top_k) if use_cache: # 将结果存入缓存 cache_file f./cache/{query_hash}.pkl os.makedirs(os.path.dirname(cache_file), exist_okTrue) with open(cache_file, wb) as f: pickle.dump(result, f) return result对于大量并发的查询可以考虑使用异步IO来并行处理检索和多个LLM调用如果一次查询需要多步LLM推理。5.2 提示词工程迭代LLM的表现极度依赖提示词。我通过多次实验总结了几个有效的提示词技巧角色扮演让LLM扮演“领域专家”如“资深技术架构师”其抽取的准确度会比通用指令更高。少样本示例Few-Shot在提示词中提供1-2个完美的输入输出示例能显著规范LLM的输出格式和质量。分步指令对于复杂查询可以要求LLM先“列出查询中涉及的核心概念”再“从文本中找出与这些概念相关的陈述”最后“将陈述转化为图谱三元组”。这比一步到位成功率更高。后处理校验LLM可能生成重复实体或矛盾关系。可以写一个简单的后处理脚本合并相同ID的实体或根据规则如“优于”和“劣于”不应同时存在于相同两个实体间进行冲突检测和清理。5.3 成本控制与监控LLM API调用是主要成本。必须进行监控和优化。import tiktoken def count_tokens(text: str, model: str gpt-4) - int: 计算文本的Token数量 encoder tiktoken.encoding_for_model(model) return len(encoder.encode(text)) # 在构图函数中记录Token消耗 def build_knowledge_graph_with_cost_tracking(query: str, retriever): # ... 检索 ... input_context context_text[:5000] # 可以截断过长的上下文 input_prompt f{schema_prompt}\n\n查询:{query}\n上下文:{input_context} input_tokens count_tokens(input_prompt, gpt-4) # 调用LLM... output_tokens count_tokens(response.content, gpt-4) total_cost (input_tokens * 0.03 output_tokens * 0.06) / 1000 # GPT-4粗略定价 print(f本次构图消耗: 输入{input_tokens} tokens, 输出{output_tokens} tokens, 估算成本${total_cost:.4f}) # 可以将消耗记录到日志或数据库 # log_cost(query, input_tokens, output_tokens, total_cost) return result_graph优化策略上下文截断只将最相关的文档片段传给LLM。使用更便宜的模型在非关键步骤如初步检索结果摘要使用GPT-3.5。设置预算和告警在应用层面设置每日/每月Token消耗上限。6. 常见问题与排查实录在实际部署和测试中我遇到了不少问题这里记录下最典型的几个及其解决方案。6.1 LLM输出格式不稳定问题LLM有时不返回JSON而是返回一段文字描述或者JSON格式错误如缺少引号。解决方案强化系统提示词在系统指令中明确强调“请输出纯净的JSON不要包含任何额外的解释或Markdown标记”。使用LangChain的Output ParsersPydanticOutputParser或JsonOutputParser可以强制LLM输出指定格式并在解析失败时进行重试或错误处理。后处理清洗如上文代码所示尝试从返回内容中提取被Markdown代码块包裹的JSON。降级模型如果GPT-4仍然不稳定可以尝试使用专门针对JSON格式进行微调的模型或者在提示词中提供更详细的JSON Schema。6.2 检索结果不相关导致构图偏差问题向量检索返回的文档片段与用户查询的意图表面相似但实际不相关导致LLM基于错误信息构图。解决方案优化检索器如前所述使用ContextualCompressionRetriever进行重排和过滤。混合检索Hybrid Search结合关键词检索如BM25和向量检索。ChromaDB支持同时进行。关键词检索能保证精确匹配向量检索保证语义匹配两者取并集或加权得分。# ChromaDB 支持传入 search_type 参数 retriever vector_store.as_retriever( search_typesimilarity, # 或 mmr (最大边际相关性) 进行多样性检索 search_kwargs{k: 6, “score_threshold”: 0.5} # 可以设置相似度阈值 )查询扩展Query Expansion在检索前先用LLM对原始查询进行改写或扩展生成多个同义或相关的查询语句分别检索后再合并结果。这能提高召回率。6.3 图谱规模失控或过于稀疏问题对于开放式查询LLM可能抽取过多无关实体导致图谱庞大且混乱或者抽取的实体和关系过少图谱没有价值。解决方案在提示词中约束范围明确要求“图谱应直接服务于回答该查询避免抽取无关的宽泛知识”。可以要求LLM先判断查询意图再决定抽取范围。设置抽取上限在提示词中要求“最多抽取5个核心实体和7条关键关系”。后处理剪枝构图后计算图中节点的度连接数过滤掉孤立节点或连接数极少的节点。或者只保留与查询中明确提到的核心实体有路径连接的子图。6.4 处理歧义与冲突信息问题不同文档可能对同一事实描述有冲突如A文档说方案X成本1000B文档说成本1200。LLM可能随机选择一个或生成矛盾关系。解决方案在提示词中要求标注来源让LLM在抽取每个事实时注明其来源于哪个文档片段通过元数据。在后端我们可以呈现“根据文档A成本为1000根据文档B成本为1200”将冲突暴露给用户判断。置信度评分可以设计简单规则如出现频率高、来源权威性高的信息置信度高。在图谱可视化中用边的粗细或颜色表示置信度。人工反馈循环允许用户对图谱中的关系进行“确认”或“纠错”将这些反馈记录下來用于优化后续的提示词或作为新的训练数据。这次Graphiti实战让我深刻体会到将LLM的动态理解能力与图谱的结构化表达能力结合是解锁非结构化数据价值的一把利器。它不像传统方法那样追求“大而全”的完美图谱而是追求“小而准”的即时洞察非常贴合快速变化的知识库和探索式问答场景。最大的挑战和乐趣都来自于与LLM的“沟通艺术”——如何通过精妙的提示词让它成为一个可靠的知识工程师。
1. 项目概述从海量文档到实时洞察的挑战最近在做一个内部项目需要把公司过去几年积累的几百份技术文档、会议纪要和产品手册“盘活”。老板的要求很直接能不能做个系统让新来的同事问个问题比如“我们A产品的数据备份方案和B方案在成本上有啥区别”系统能立刻从这些文档里找到相关信息并且把产品、方案、成本这些概念之间的关系清晰地展示出来而不是扔给用户一堆需要自己再整理的搜索结果。这个需求本质上就是构建一个能够实时查询和推理的知识图谱系统。传统的知识图谱构建往往是个“离线批处理”的活儿。你需要用NLP工具从文档里抽取出实体比如产品名、技术名词和关系比如“包含”、“优于”、“依赖于”然后存进图数据库。这个过程耗时很长数据更新也不及时。而“实时”的要求意味着我们需要一种更敏捷的方式当用户提出一个新问题时系统能动态地从最新的文档中抽取知识并即时构建出一个针对该问题的、轻量级的图谱片段进行展示和推理。这就是我选择Graphiti这个框架进行实战探索的核心原因。Graphiti 并不是一个图数据库而是一个将大语言模型LLM的语义理解能力与图结构Graph的关联推理能力结合起来的开发框架。它的核心思路是“按需构图”。系统不会事先把所有可能的知识都抽取并存储成一张大图那成本高且维护难而是利用LLM理解用户查询的意图动态地决定需要从文档中抽取哪些实体和关系并即时组织成图谱进行回答。这对于处理海量、非结构化文档且需求多变的场景来说非常具有吸引力。接下来我将完整分享这次实战的笔记包括设计思路、关键实现、踩过的坑以及一些性能调优的心得。2. 核心架构与工具选型解析2.1 为什么是 Graphiti LLM 向量数据库的组合面对“海量文档实时知识图谱”的需求我评估了几个方案。传统方案是NLP流水线实体识别、关系抽取 图数据库Neo4j, NebulaGraph。这个方案的问题是流水线需要大量标注数据来训练或者依赖规则泛化能力差且构建的是“静态全图”任何文档更新都需要重新跑一遍流程无法实时。Graphiti提出的动态构图理念更符合我们的场景。其架构核心是LLM作为“图谱构建师”利用LLM强大的零样本/少样本理解能力将用户查询和文档片段转化为图谱查询指令或直接生成图谱结构。它替代了传统的训练好的NLP模型。向量数据库作为“记忆库”所有文档被切分成片段chunks编码成向量后存入向量数据库如Chroma, Weaviate。当用户提问时先将问题本身也转化为向量在向量库中进行相似性检索快速找到最相关的文档片段。这解决了“从海量文档中快速定位相关信息”的问题。Graphiti作为“协调中枢”Graphiti框架负责编排整个流程。它接收用户查询调用LLM分析查询意图并生成针对向量检索结果的“信息抽取指令”然后再调用LLM根据指令和检索到的文本抽取出实体和关系最后组织成图结构返回。我最终的技术栈如下框架GraphitiPythonLLM服务OpenAI GPT-4 API用于复杂意图理解和信息抽取。对于成本敏感的部分也用到了 GPT-3.5-Turbo。向量数据库ChromaDB轻量级易于集成适合原型和中小规模数据。开发语言Python 3.10。辅助工具LangChain用于文档加载、文本分割和部分链式调用编排但Graphiti本身也提供了类似的模式。注意LLM API的选择至关重要。GPT-4在理解复杂指令和进行精确抽取方面显著优于3.5但成本也高。我的策略是在构图的关键步骤如解析查询生成抽取指令使用GPT-4以保证质量在简单的文本摘要或初筛时使用GPT-3.5。2.2 环境准备与初始化首先建立一个干净的Python环境并安装核心依赖。# 创建并激活虚拟环境 python -m venv graphiti_env source graphiti_env/bin/activate # Linux/Mac # graphiti_env\Scripts\activate # Windows # 安装核心包 pip install graphiti-ai pip install openai pip install chromadb pip install langchain langchain-openai pip install pypdf # 用于读取PDF文档 pip install tiktoken # 用于计算Token控制成本接下来进行关键的初始化配置主要是设置LLM和向量数据库。import os from graphiti import Graphiti from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain_community.vectorstores import Chroma from langchain_text_splitters import RecursiveCharacterTextSplitter # 1. 设置OpenAI API密钥请替换为你的密钥或从环境变量读取 os.environ[OPENAI_API_KEY] your-api-key-here # 2. 初始化LLM客户端 # 用于对话和复杂推理的LLM llm_gpt4 ChatOpenAI(modelgpt-4, temperature0.1) # temperature调低使输出更确定 llm_gpt35 ChatOpenAI(modelgpt-3.5-turbo, temperature0.1) # 用于生成文本嵌入向量的模型 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 性价比高 # 3. 初始化Graphiti graphiti_client Graphiti() # 4. 初始化文本分割器 # 这里选择递归字符分割尝试保持段落和句子的完整性 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, # 每个片段约1000字符 chunk_overlap200, # 片段间重叠200字符避免信息被割裂 length_functionlen, separators[\n\n, \n, 。, , , , , , ] )实操心得chunk_size的设置是平衡检索精度和上下文完整性的关键。太小如200会导致信息碎片化LLM缺乏足够上下文进行准确抽取太大如2000则可能引入无关噪声降低向量检索的准确性。1000-1500是一个常见的起步值需要根据你的文档平均段落长度进行调整。chunk_overlap能有效缓解句子被腰斩的问题。3. 知识库构建文档处理与向量化知识图谱的“知识”来源于文档因此第一步是将非结构化的文档转化为结构化的、可检索的知识单元。3.1 文档加载与预处理我处理的文档包括PDF、Word和Markdown。使用LangChain的文档加载器可以统一处理。from langchain_community.document_loaders import PyPDFLoader, UnstructuredWordDocumentLoader, TextLoader from typing import List def load_documents(directory_path: str) - List[Document]: 加载指定目录下的所有支持格式的文档 documents [] for filename in os.listdir(directory_path): filepath os.path.join(directory_path, filename) if filename.endswith(.pdf): loader PyPDFLoader(filepath) elif filename.endswith(.docx): loader UnstructuredWordDocumentLoader(filepath) elif filename.endswith(.md) or filename.endswith(.txt): loader TextLoader(filepath, encodingutf-8) else: continue loaded_docs loader.load() # 为每个文档片段添加源文件元数据便于追溯 for doc in loaded_docs: doc.metadata[source] filename documents.extend(loaded_docs) return documents # 示例加载docs文件夹下的所有文档 raw_documents load_documents(./docs) print(f共加载了 {len(raw_documents)} 个原始文档片段。)3.2 文本分割与向量数据库持久化加载后的文档需要被分割成更小的片段然后转化为向量存入ChromaDB。def create_vector_store(documents: List[Document], persist_directory: str ./chroma_db): 将文档分割、向量化并持久化到ChromaDB。 # 1. 分割文本 print(开始分割文本...) all_splits text_splitter.split_documents(documents) print(f分割后得到 {len(all_splits)} 个文本片段。) # 2. 创建并持久化向量存储 print(开始生成向量并存入数据库...) vectorstore Chroma.from_documents( documentsall_splits, embeddingembeddings, persist_directorypersist_directory ) # 显式持久化 vectorstore.persist() print(f向量数据库已创建并保存至 {persist_directory}) return vectorstore # 执行创建 vector_store create_vector_store(raw_documents)注意事项向量数据库的持久化路径很重要。首次运行会创建后续运行可以直接加载无需重复向量化节省成本和时间。# 后续加载已有向量数据库 vector_store Chroma(persist_directory./chroma_db, embedding_functionembeddings)3.3 检索策略优化提升召回率简单的向量相似性检索有时会漏掉关键信息。我采用了混合检索策略来提升召回率。from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor def create_enhanced_retriever(vectorstore, llm_for_compressionllm_gpt35): 创建增强的检索器结合向量检索和上下文压缩。 # 基础向量检索器设置检索数量稍大 base_retriever vector_store.as_retriever(search_kwargs{k: 8}) # 使用LLM对检索结果进行压缩/重排只保留与查询最相关的部分 # 这可以节省后续LLM处理的Token并提升精度 compressor LLMChainExtractor.from_llm(llm_for_compression) compression_retriever ContextualCompressionRetriever( base_compressorcompressor, base_retrieverbase_retriever ) return compression_retriever enhanced_retriever create_enhanced_retriever(vector_store)这个LLMChainExtractor会在向量检索返回片段后再用一个小型LLM我用GPT-3.5快速扫描每个片段提取出其中与查询直接相关的句子过滤掉无关内容。实测下来这能有效提高最终构图时输入LLM的上下文质量。4. 动态图谱构建Graphiti 核心实战这是最核心的部分即如何利用Graphiti和LLM根据用户查询动态构建知识图谱。4.1 定义图谱模式Schema虽然我们是动态构图但提前定义一个期望的图谱模式能极大引导LLM的输出格式使其更结构化、更可控。我们定义实体类型和关系类型。# 根据我们的技术文档领域定义可能的实体和关系类型 GRAPH_SCHEMA { entity_types: [ 产品, 技术组件, 功能特性, 部署环境, 人员角色, 成本指标, 问题, 解决方案 ], relationship_types: [ 包含, 依赖于, 优于, 劣于, 导致, 解决, 拥有, 属于, 相关于, 成本为 ] } # 将这个模式转化为给LLM的提示词部分 schema_prompt f 你是一个知识图谱构建专家。请从给定的文本中提取信息构建一个知识图谱。 图谱中的节点实体类型应限于{, .join(GRAPH_SCHEMA[entity_types])}。 图谱中的边关系类型应限于{, .join(GRAPH_SCHEMA[relationship_types])}。 请以JSON格式输出包含entities和relationships两个列表。 每个实体应包含id唯一标识如‘产品_A’、name显示名称、type实体类型。 每个关系应包含source_id源实体ID、target_id目标实体ID、type关系类型、description可选关系描述。 4.2 实现动态构图管道现在我们将检索、LLM调用和结果解析串联起来。import json from langchain_core.prompts import ChatPromptTemplate def build_knowledge_graph(query: str, retriever, top_k: int 5) - dict: 核心函数根据用户查询动态构建知识图谱。 # 1. 检索相关文档片段 print(f检索与查询‘{query}’相关的文档...) relevant_docs retriever.invoke(query) context_text \n\n---\n\n.join([doc.page_content for doc in relevant_docs]) if not context_text: return {entities: [], relationships: [], context: 未找到相关信息。} # 2. 构建LLM提示词 prompt_template ChatPromptTemplate.from_messages([ (system, 你是一个精准的信息抽取助手。 schema_prompt), (human, 用户查询{query} 请基于以下相关文本内容抽取与查询意图紧密相关的实体和关系构建一个聚焦的知识图谱。 注意图谱应直接服务于回答该查询避免抽取无关的宽泛知识。 相关文本 {context} 请输出纯净的JSON。 ) ]) # 3. 调用LLM使用GPT-4以保证抽取质量 chain prompt_template | llm_gpt4 response chain.invoke({query: query, context: context_text}) # 4. 解析LLM的JSON输出 try: # LLM输出可能是带Markdown代码块的JSON需要清理 content response.content if json in content: content content.split(json)[1].split()[0].strip() elif in content: content content.split()[1].split()[0].strip() graph_data json.loads(content) # 添加检索到的源文档信息作为图谱元数据 graph_data[source_documents] [doc.metadata.get(source, unknown) for doc in relevant_docs] return graph_data except json.JSONDecodeError as e: print(fLLM返回的JSON解析失败: {e}) print(f原始返回内容: {response.content}) # 返回一个包含错误信息的空结构 return {entities: [], relationships: [], error: 图谱生成失败LLM返回格式异常。} # 示例查询 query_example “我们产品A的数据备份方案和产品B的容灾方案在成本上有何差异” result_graph build_knowledge_graph(query_example, enhanced_retriever) print(json.dumps(result_graph, indent2, ensure_asciiFalse))一个成功的输出可能如下所示{ entities: [ {id: 产品_A, name: 产品A, type: 产品}, {id: 备份方案_X, name: 基于快照的增量备份方案, type: 解决方案}, {id: 成本_月度1000, name: 每月1000元, type: 成本指标}, {id: 产品_B, name: 产品B, type: 产品}, {id: 容灾方案_Y, name: 跨地域热备容灾方案, type: 解决方案}, {id: 成本_月度5000, name: 每月5000元, type: 成本指标} ], relationships: [ {source_id: 产品_A, target_id: 备份方案_X, type: 拥有, description: 采用}, {source_id: 备份方案_X, target_id: 成本_月度1000, type: 成本为, description: null}, {source_id: 产品_B, target_id: 容灾方案_Y, type: 拥有, description: 采用}, {source_id: 容灾方案_Y, target_id: 成本_月度5000, type: 成本为, description: null}, {source_id: 备份方案_X, target_id: 容灾方案_Y, type: 劣于, description: 在容灾级别和成本上} ], source_documents: [产品A白皮书.pdf, 产品B技术架构.docx, 成本核算指南.md] }4.3 图谱可视化与交互生成JSON数据后我们可以用网络图库进行可视化。这里使用networkx和pyvis。import networkx as nx from pyvis.network import Network def visualize_graph(graph_data: dict, output_html_path: str knowledge_graph.html): 将图谱数据可视化为交互式HTML网页。 G nx.DiGraph() # 创建有向图 # 添加节点 for entity in graph_data.get(entities, []): G.add_node( entity[id], labelentity[name], titlef类型: {entity[type]}, groupentity[type] # 按类型分组便于可视化区分 ) # 添加边 for rel in graph_data.get(relationships, []): G.add_edge( rel[source_id], rel[target_id], titlerel[type] (f: {rel[description]} if rel.get(description) else ), labelrel[type] ) # 使用Pyvis生成交互式网络 net Network(height750px, width100%, directedTrue, notebookFalse) net.from_nx(G) # 可以调整一些物理布局参数让图更美观 net.set_options( var options { physics: { forceAtlas2Based: { gravitationalConstant: -50, centralGravity: 0.01, springLength: 100, springConstant: 0.08 }, minVelocity: 0.75, solver: forceAtlas2Based } } ) net.save_graph(output_html_path) print(f知识图谱已可视化保存至: {output_html_path}) # 在Jupyter Notebook中可以直接显示 # return net.show(f{output_html_path}) # 可视化上面生成的图谱 visualize_graph(result_graph)生成的HTML文件可以在浏览器中打开你可以拖动节点放大缩小清晰地看到“产品A-拥有-备份方案X-成本为-月度1000元”以及“劣于”关系连接到产品B的容灾方案。这种可视化对于呈现复杂关系非常直观。5. 高级技巧与性能优化5.1 缓存与异步处理频繁调用LLM API成本高、速度慢。对于相对稳定的文档库可以对“查询-检索结果-构图”进行缓存。import hashlib import pickle from functools import lru_cache def get_query_hash(query: str, top_k: int) - str: 生成查询的哈希键用于缓存 return hashlib.md5(f{query}_{top_k}.encode()).hexdigest() lru_cache(maxsize100) def cached_build_graph(query_hash: str, context_text: str) - dict: 缓存构图结果。注意context_text是检索到的文本如果文档库更新缓存需要失效。 这里为简化假设文档库短期内不变。生产环境需用Redis等外部缓存并设置过期。 # 这里模拟一个缓存查找实际应连接缓存数据库 cache_file f./cache/{query_hash}.pkl if os.path.exists(cache_file): with open(cache_file, rb) as f: return pickle.load(f) # 如果没有缓存则调用真正的构图函数这里需要原函数支持略作修改 # 实际应用中应将构图逻辑封装此处仅为示意 return None # 修改后的构图函数加入缓存逻辑 def build_knowledge_graph_with_cache(query: str, retriever, top_k: int 5, use_cacheTrue) - dict: query_hash get_query_hash(query, top_k) if use_cache: cached_result cached_build_graph(query_hash, ) # 需要更精细的缓存键设计 if cached_result: print(命中缓存) return cached_result # ... (原有的检索和构图逻辑) ... result build_knowledge_graph(query, retriever, top_k) if use_cache: # 将结果存入缓存 cache_file f./cache/{query_hash}.pkl os.makedirs(os.path.dirname(cache_file), exist_okTrue) with open(cache_file, wb) as f: pickle.dump(result, f) return result对于大量并发的查询可以考虑使用异步IO来并行处理检索和多个LLM调用如果一次查询需要多步LLM推理。5.2 提示词工程迭代LLM的表现极度依赖提示词。我通过多次实验总结了几个有效的提示词技巧角色扮演让LLM扮演“领域专家”如“资深技术架构师”其抽取的准确度会比通用指令更高。少样本示例Few-Shot在提示词中提供1-2个完美的输入输出示例能显著规范LLM的输出格式和质量。分步指令对于复杂查询可以要求LLM先“列出查询中涉及的核心概念”再“从文本中找出与这些概念相关的陈述”最后“将陈述转化为图谱三元组”。这比一步到位成功率更高。后处理校验LLM可能生成重复实体或矛盾关系。可以写一个简单的后处理脚本合并相同ID的实体或根据规则如“优于”和“劣于”不应同时存在于相同两个实体间进行冲突检测和清理。5.3 成本控制与监控LLM API调用是主要成本。必须进行监控和优化。import tiktoken def count_tokens(text: str, model: str gpt-4) - int: 计算文本的Token数量 encoder tiktoken.encoding_for_model(model) return len(encoder.encode(text)) # 在构图函数中记录Token消耗 def build_knowledge_graph_with_cost_tracking(query: str, retriever): # ... 检索 ... input_context context_text[:5000] # 可以截断过长的上下文 input_prompt f{schema_prompt}\n\n查询:{query}\n上下文:{input_context} input_tokens count_tokens(input_prompt, gpt-4) # 调用LLM... output_tokens count_tokens(response.content, gpt-4) total_cost (input_tokens * 0.03 output_tokens * 0.06) / 1000 # GPT-4粗略定价 print(f本次构图消耗: 输入{input_tokens} tokens, 输出{output_tokens} tokens, 估算成本${total_cost:.4f}) # 可以将消耗记录到日志或数据库 # log_cost(query, input_tokens, output_tokens, total_cost) return result_graph优化策略上下文截断只将最相关的文档片段传给LLM。使用更便宜的模型在非关键步骤如初步检索结果摘要使用GPT-3.5。设置预算和告警在应用层面设置每日/每月Token消耗上限。6. 常见问题与排查实录在实际部署和测试中我遇到了不少问题这里记录下最典型的几个及其解决方案。6.1 LLM输出格式不稳定问题LLM有时不返回JSON而是返回一段文字描述或者JSON格式错误如缺少引号。解决方案强化系统提示词在系统指令中明确强调“请输出纯净的JSON不要包含任何额外的解释或Markdown标记”。使用LangChain的Output ParsersPydanticOutputParser或JsonOutputParser可以强制LLM输出指定格式并在解析失败时进行重试或错误处理。后处理清洗如上文代码所示尝试从返回内容中提取被Markdown代码块包裹的JSON。降级模型如果GPT-4仍然不稳定可以尝试使用专门针对JSON格式进行微调的模型或者在提示词中提供更详细的JSON Schema。6.2 检索结果不相关导致构图偏差问题向量检索返回的文档片段与用户查询的意图表面相似但实际不相关导致LLM基于错误信息构图。解决方案优化检索器如前所述使用ContextualCompressionRetriever进行重排和过滤。混合检索Hybrid Search结合关键词检索如BM25和向量检索。ChromaDB支持同时进行。关键词检索能保证精确匹配向量检索保证语义匹配两者取并集或加权得分。# ChromaDB 支持传入 search_type 参数 retriever vector_store.as_retriever( search_typesimilarity, # 或 mmr (最大边际相关性) 进行多样性检索 search_kwargs{k: 6, “score_threshold”: 0.5} # 可以设置相似度阈值 )查询扩展Query Expansion在检索前先用LLM对原始查询进行改写或扩展生成多个同义或相关的查询语句分别检索后再合并结果。这能提高召回率。6.3 图谱规模失控或过于稀疏问题对于开放式查询LLM可能抽取过多无关实体导致图谱庞大且混乱或者抽取的实体和关系过少图谱没有价值。解决方案在提示词中约束范围明确要求“图谱应直接服务于回答该查询避免抽取无关的宽泛知识”。可以要求LLM先判断查询意图再决定抽取范围。设置抽取上限在提示词中要求“最多抽取5个核心实体和7条关键关系”。后处理剪枝构图后计算图中节点的度连接数过滤掉孤立节点或连接数极少的节点。或者只保留与查询中明确提到的核心实体有路径连接的子图。6.4 处理歧义与冲突信息问题不同文档可能对同一事实描述有冲突如A文档说方案X成本1000B文档说成本1200。LLM可能随机选择一个或生成矛盾关系。解决方案在提示词中要求标注来源让LLM在抽取每个事实时注明其来源于哪个文档片段通过元数据。在后端我们可以呈现“根据文档A成本为1000根据文档B成本为1200”将冲突暴露给用户判断。置信度评分可以设计简单规则如出现频率高、来源权威性高的信息置信度高。在图谱可视化中用边的粗细或颜色表示置信度。人工反馈循环允许用户对图谱中的关系进行“确认”或“纠错”将这些反馈记录下來用于优化后续的提示词或作为新的训练数据。这次Graphiti实战让我深刻体会到将LLM的动态理解能力与图谱的结构化表达能力结合是解锁非结构化数据价值的一把利器。它不像传统方法那样追求“大而全”的完美图谱而是追求“小而准”的即时洞察非常贴合快速变化的知识库和探索式问答场景。最大的挑战和乐趣都来自于与LLM的“沟通艺术”——如何通过精妙的提示词让它成为一个可靠的知识工程师。
