当前位置：首页 > news >正文

4 LlamaIndex 全流程实践：构建企业级智能文档问答系统

news 2025/11/5 16:26:47

LlamaIndex 全流程实践：构建企业级智能文档问答系统

引言

在企业环境中，大量的知识以非结构化文档的形式存在，如PDF报告、Word文档、网页内容等。LlamaIndex（原名GPT Index）是一个专门用于构建基于私有数据的智能问答系统的框架。本节将深入探讨LlamaIndex的全流程实践，从文档加载到智能问答，构建完整的企业级知识增强应用。

LlamaIndex 架构概览

首先让我们通过架构图理解LlamaIndex的核心组件和工作流程：

文档加载器（Document Loaders）

LlamaIndex支持多种格式的文档加载，让我们从基础开始：

基础文档加载

import os
from llama_index import SimpleDirectoryReader, Document
from llama_index.node_parser import SimpleNodeParser
from llama_index.schema import IndexNode
import requests
from pathlib import Pathclass DocumentLoaderDemo:"""文档加载器演示"""def __init__(self, data_dir: str = "./data"):"""初始化文档加载器Args:data_dir: 数据目录路径"""self.data_dir = Path(data_dir)self.data_dir.mkdir(exist_ok=True)def demo_simple_loader(self):"""演示简单目录加载器"""print("=== 简单目录加载器演示 ===")# 创建示例文档sample_docs = [("公司介绍.md", """# 公司介绍## 公司概况我们是一家专注于人工智能技术的创新公司，成立于2020年。## 核心业务- AI技术咨询- 智能客服系统开发- 企业数字化转型解决方案## 联系方式电话: 400-123-4567邮箱: contact@company.com"""),("产品服务.md", """# 产品与服务## 智能客服系统基于大语言模型的智能客服系统，支持多轮对话和意图识别。### 主要功能1. 自动问答2. 工单管理3. 数据分析## AI技术咨询为企业提供AI技术落地咨询服务。"""),("技术文档.md", """# 技术架构文档## 系统架构我们的系统采用微服务架构，包含以下组件：- API网关- 用户服务- 订单服务- 支付服务## 技术栈- 后端: Python, FastAPI- 前端: React, TypeScript- 数据库: PostgreSQL, Redis""")]# 保存示例文档for filename, content in sample_docs:file_path = self.data_dir / filenamewith open(file_path, 'w', encoding='utf-8') as f:f.write(content)print(f"创建文档: {filename}")# 使用SimpleDirectoryReader加载文档reader = SimpleDirectoryReader(input_dir=str(self.data_dir))documents = reader.load_data()print(f"\n成功加载 {len(documents)} 个文档:")for i, doc in enumerate(documents):print(f"文档 {i+1}: {len(doc.text)} 字符")print(f"前100字符: {doc.text[:100]}...")print("-" * 50)return documentsdef demo_custom_document_creation(self):"""演示自定义文档创建"""print("\n=== 自定义文档创建 ===")# 创建自定义文档custom_documents = [Document(text="""人工智能发展趋势报告 2024主要趋势：1. 多模态大模型成为主流2. AI Agent技术快速发展3. 边缘AI部署增加4. AI安全与伦理受到重视""",metadata={"source": "内部报告","author": "技术研究部","date": "2024-01-15","category": "技术趋势"}),Document(text="""季度业务总结 Q1 2024业绩亮点：- 收入增长25%- 客户数量增加40%- 新产品线贡献30%收入下季度目标：- 扩大市场份额- 优化客户体验- 推出AI增强功能""",metadata={"source": "业务报告", "author": "业务发展部","date": "2024-03-31","category": "业务报告"})]print("自定义文档信息:")for i, doc in enumerate(custom_documents):print(f"文档 {i+1}:")print(f"  内容长度: {len(doc.text)} 字符")print(f"  元数据: {doc.metadata}")print(f"  前80字符: {doc.text[:80]}...")print("-" * 50)return custom_documents# 高级文档加载器
class AdvancedDocumentLoader:"""高级文档加载器"""def __init__(self):passdef load_pdf_documents(self, pdf_path: str):"""加载PDF文档"""try:from llama_index import SimpleDirectoryReader# 配置PDF加载器reader = SimpleDirectoryReader(input_files=[pdf_path],required_exts=[".pdf"])documents = reader.load_data()return documentsexcept ImportError:print("请安装PDF支持: pip install pypdf")return []def load_web_content(self, urls: list):"""加载网页内容"""try:from llama_index import SimpleDirectoryReaderfrom llama_index.readers.web import BeautifulSoupWebReader# 使用BeautifulSoupWebReaderloader = BeautifulSoupWebReader()documents = loader.load_data(urls=urls)return documentsexcept Exception as e:print(f"网页加载失败: {e}")return []def load_from_database(self, query: str, connection_string: str):"""从数据库加载数据"""# 模拟数据库数据sample_data = [{"id": 1, "question": "如何重置密码?", "answer": "在设置页面点击忘记密码，按照指引操作。"},{"id": 2, "question": "如何联系客服?", "answer": "拨打400-123-4567或发送邮件至support@company.com。"},{"id": 3, "question": "服务费用是多少?", "answer": "基础版免费，专业版每月299元，企业版请联系销售。"}]documents = []for item in sample_data:text = f"问题: {item['question']}\n答案: {item['answer']}"doc = Document(text=text,metadata={"source": "知识库","id": item["id"],"type": "FAQ"})documents.append(doc)return documents# 运行文档加载演示
def run_document_loader_demos():"""运行文档加载演示"""print("开始文档加载演示...")# 基础加载器演示loader_demo = DocumentLoaderDemo()documents = loader_demo.demo_simple_loader()custom_docs = loader_demo.demo_custom_document_creation()# 高级加载器演示advanced_loader = AdvancedDocumentLoader()# 演示数据库文档加载print("\n=== 数据库文档加载演示 ===")db_docs = advanced_loader.load_from_database("SELECT * FROM faq", "sqlite:///sample.db")print(f"从数据库加载了 {len(db_docs)} 个FAQ文档")return documents + custom_docs + db_docsif __name__ == "__main__":all_documents = run_document_loader_demos()

索引构建（Indexing）

索引是LlamaIndex的核心，它决定了文档的存储和检索方式。

基础索引构建

from llama_index import VectorStoreIndex, ServiceContext
from llama_index.llms import OpenAI
from llama_index.embeddings import OpenAIEmbedding
from llama_index.node_parser import SimpleNodeParser, HierarchicalNodeParser
import chromadb
from llama_index.vector_stores import ChromaVectorStoreclass IndexBuilder:"""索引构建器"""def __init__(self, openai_api_key: str):"""初始化索引构建器Args:openai_api_key: OpenAI API密钥"""self.llm = OpenAI(api_key=openai_api_key, model="gpt-3.5-turbo")self.embed_model = OpenAIEmbedding(api_key=openai_api_key)# 创建服务上下文self.service_context = ServiceContext.from_defaults(llm=self.llm,embed_model=self.embed_model,chunk_size=512,chunk_overlap=50)def build_basic_index(self, documents):"""构建基础向量索引"""print("=== 构建基础向量索引 ===")# 创建节点解析器node_parser = SimpleNodeParser.from_defaults(chunk_size=512,chunk_overlap=50)# 解析文档为节点nodes = node_parser.get_nodes_from_documents(documents)print(f"将文档解析为 {len(nodes)} 个节点")# 构建向量索引index = VectorStoreIndex(nodes=nodes,service_context=self.service_context)return indexdef build_chroma_index(self, documents, persist_dir: str = "./chroma_db"):"""构建Chroma向量存储索引"""print("\n=== 构建Chroma向量存储索引 ===")# 初始化Chroma客户端chroma_client = chromadb.PersistentClient(path=persist_dir)chroma_collection = chroma_client.get_or_create_collection("knowledge_base")# 创建向量存储vector_store = ChromaVectorStore(chroma_collection=chroma_collection)# 构建索引index = VectorStoreIndex.from_documents(documents,vector_store=vector_store,service_context=self.service_context)print(f"索引已保存到: {persist_dir}")return indexdef build_hierarchical_index(self, documents):"""构建分层索引"""print("\n=== 构建分层索引 ===")# 使用分层节点解析器node_parser = HierarchicalNodeParser.from_defaults(chunk_sizes=[1024, 512, 256])# 解析文档nodes = node_parser.get_nodes_from_documents(documents)print(f"分层解析创建了 {len(nodes)} 个节点")# 构建索引index = VectorStoreIndex(nodes, service_context=self.service_context)return index# 高级索引策略
class AdvancedIndexingStrategies:"""高级索引策略"""def __init__(self, openai_api_key: str):self.service_context = ServiceContext.from_defaults(llm=OpenAI(api_key=openai_api_key),embed_model=OpenAIEmbedding(api_key=openai_api_key))def create_summary_index(self, documents):"""创建摘要索引"""from llama_index import SummaryIndexprint("=== 创建摘要索引 ===")index = SummaryIndex.from_documents(documents,service_context=self.service_context)return indexdef create_keyword_table_index(self, documents):"""创建关键词表索引"""from llama_index import KeywordTableIndexprint("=== 创建关键词表索引 ===")index = KeywordTableIndex.from_documents(documents,service_context=self.service_context)return indexdef create_composite_index(self, documents):"""创建复合索引"""from llama_index import ComposableGraphprint("=== 创建复合索引 ===")# 创建不同类型的子索引vector_index = VectorStoreIndex.from_documents(documents, service_context=self.service_context)summary_index = SummaryIndex.from_documents(documents, service_context=self.service_context)# 构建复合索引graph = ComposableGraph.from_indices(SummaryIndex,[vector_index, summary_index],index_summaries=["向量索引", "摘要索引"],service_context=self.service_context)return graph# 运行索引构建演示
def run_indexing_demos(documents, openai_api_key: str):"""运行索引构建演示"""print("开始索引构建演示...")# 基础索引构建index_builder = IndexBuilder(openai_api_key)basic_index = index_builder.build_basic_index(documents)# Chroma索引构建chroma_index = index_builder.build_chroma_index(documents)# 分层索引构建hierarchical_index = index_builder.build_hierarchical_index(documents)# 高级索引策略advanced_strategies = AdvancedIndexingStrategies(openai_api_key)summary_index = advanced_strategies.create_summary_index(documents)keyword_index = advanced_strategies.create_keyword_table_index(documents)composite_index = advanced_strategies.create_composite_index(documents)return {"basic": basic_index,"chroma": chroma_index,"hierarchical": hierarchical_index,"summary": summary_index,"keyword": keyword_index,"composite": composite_index}# 索引性能分析
class IndexAnalyzer:"""索引分析器"""def analyze_index(self, index, index_name: str):"""分析索引性能"""print(f"\n=== {index_name} 索引分析 ===")if hasattr(index, 'index_struct'):struct = index.index_structprint(f"索引类型: {type(struct).__name__}")if hasattr(struct, 'nodes'):print(f"节点数量: {len(struct.nodes)}")if hasattr(struct, 'summary_dict'):print(f"摘要数量: {len(struct.summary_dict)}")# 测试查询性能import timestart_time = time.time()# 执行简单查询测试query_engine = index.as_query_engine()test_response = query_engine.query("测试查询")query_time = time.time() - start_timeprint(f"查询响应时间: {query_time:.2f}秒")print(f"响应长度: {len(str(test_response))} 字符")if __name__ == "__main__":# 加载文档documents = run_document_loader_demos()# 构建索引openai_api_key = os.getenv("OPENAI_API_KEY")if openai_api_key:indexes = run_indexing_demos(documents, openai_api_key)# 分析索引性能analyzer = IndexAnalyzer()for name, index in indexes.items():analyzer.analyze_index(index, name)

查询引擎（Query Engine）

查询引擎是用户与索引交互的接口，支持多种查询策略。

基础查询引擎

from llama_index import QueryBundle
from llama_index.schema import NodeWithScore
from llama_index.retrievers import VectorIndexRetriever
from llama_index.query_engine import RetrieverQueryEngine
from llama_index.response_synthesizers import get_response_synthesizerclass QueryEngineDemo:"""查询引擎演示"""def __init__(self, indexes: dict):"""初始化查询引擎演示Args:indexes: 各种类型的索引字典"""self.indexes = indexesself.setup_query_engines()def setup_query_engines(self):"""设置各种查询引擎"""self.query_engines = {}# 基础向量查询引擎if "basic" in self.indexes:self.query_engines["basic"] = self.indexes["basic"].as_query_engine(similarity_top_k=3,response_mode="compact")# 摘要索引查询引擎if "summary" in self.indexes:self.query_engines["summary"] = self.indexes["summary"].as_query_engine(response_mode="tree_summarize")# 关键词索引查询引擎if "keyword" in self.indexes:self.query_engines["keyword"] = self.indexes["keyword"].as_query_engine()# 自定义检索器查询引擎if "basic" in self.indexes:self.setup_custom_query_engine()def setup_custom_query_engine(self):"""设置自定义查询引擎"""# 创建自定义检索器retriever = VectorIndexRetriever(index=self.indexes["basic"],similarity_top_k=5)# 创建响应合成器response_synthesizer = get_response_synthesizer(response_mode="compact")# 创建自定义查询引擎custom_query_engine = RetrieverQueryEngine(retriever=retriever,response_synthesizer=response_synthesizer)self.query_engines["custom"] = custom_query_enginedef demo_basic_queries(self):"""演示基础查询"""print("=== 基础查询演示 ===")test_queries = ["公司的主要业务是什么？","智能客服系统有哪些功能？","技术架构包含哪些组件？","如何联系客服？"]for query in test_queries:print(f"\n查询: {query}")for engine_name, query_engine in self.query_engines.items():if engine_name != "custom":  # 先不测试自定义引擎try:response = query_engine.query(query)print(f"{engine_name} 引擎回复: {response}")except Exception as e:print(f"{engine_name} 引擎错误: {e}")print("-" * 60)def demo_advanced_queries(self):"""演示高级查询功能"""print("\n=== 高级查询演示 ===")# 使用自定义查询引擎进行详细分析if "custom" in self.query_engines:complex_queries = ["详细分析公司的技术栈和业务关系","总结所有文档中提到的服务类型","比较不同文档中的技术架构描述"]for query in complex_queries:print(f"\n复杂查询: {query}")response = self.query_engines["custom"].query(query)print(f"回复: {response}")print("=" * 80)# 高级查询策略
class AdvancedQueryStrategies:"""高级查询策略"""def __init__(self, index):self.index = indexdef create_sub_question_query_engine(self):"""创建子问题查询引擎"""from llama_index.query_engine import SubQuestionQueryEnginefrom llama_index.tools import QueryEngineTool# 创建查询引擎工具query_engine_tools = [QueryEngineTool(query_engine=self.index.as_query_engine(),metadata=ToolMetadata(name="知识库查询",description="用于查询公司知识库信息"))]# 创建子问题查询引擎query_engine = SubQuestionQueryEngine.from_defaults(query_engine_tools=query_engine_tools)return query_enginedef create_router_query_engine(self):"""创建路由查询引擎"""from llama_index.query_engine import RouterQueryEnginefrom llama_index.selectors.llm_selectors import LLMSingleSelectorfrom llama_index.tools import QueryEngineTool# 创建不同的查询引擎vector_query_engine = self.index.as_query_engine(similarity_top_k=3)summary_query_engine = self.index.as_query_engine(response_mode="tree_summarize")# 创建工具tools = [QueryEngineTool(query_engine=vector_query_engine,metadata=ToolMetadata(name="向量检索",description="适合具体事实检索和细节查询")),QueryEngineTool(query_engine=summary_query_engine,metadata=ToolMetadata(name="摘要生成", description="适合总结性问题和概述查询"))]# 创建路由查询引擎query_engine = RouterQueryEngine(selector=LLMSingleSelector.from_defaults(),query_engine_tools=tools)return query_engine# 查询结果分析器
class QueryResultAnalyzer:"""查询结果分析器"""def analyze_response(self, response, query: str):"""分析查询响应"""print(f"\n=== 查询分析: {query} ===")# 分析源节点if hasattr(response, 'source_nodes'):print(f"检索到 {len(response.source_nodes)} 个相关节点:")for i, node in enumerate(response.source_nodes[:3]):  # 显示前3个if hasattr(node, 'score'):print(f"节点 {i+1} (相似度: {node.score:.3f}):")else:print(f"节点 {i+1}:")# 显示节点内容预览content_preview = node.node.text[:200] + "..." if len(node.node.text) > 200 else node.node.textprint(f"  内容: {content_preview}")# 显示元数据if hasattr(node.node, 'metadata'):print(f"  元数据: {node.node.metadata}")# 显示响应内容print(f"\n最终响应: {response}")# 运行查询引擎演示
def run_query_engine_demos(indexes, openai_api_key: str):"""运行查询引擎演示"""print("开始查询引擎演示...")# 基础查询演示query_demo = QueryEngineDemo(indexes)query_demo.demo_basic_queries()query_demo.demo_advanced_queries()# 高级查询策略演示if "basic" in indexes:advanced_strategies = AdvancedQueryStrategies(indexes["basic"])print("\n=== 子问题查询引擎演示 ===")sub_question_engine = advanced_strategies.create_sub_question_query_engine()complex_query = "分析公司的业务模式和技术架构之间的关系"response = sub_question_engine.query(complex_query)print(f"复杂查询: {complex_query}")print(f"子问题引擎回复: {response}")print("\n=== 路由查询引擎演示 ===")router_engine = advanced_strategies.create_router_query_engine()test_queries = ["公司的具体技术栈是什么？",  # 应该使用向量检索"总结公司的主要业务"  # 应该使用摘要生成]for query in test_queries:response = router_engine.query(query)print(f"查询: {query}")print(f"路由引擎回复: {response}")print("-" * 60)# 查询结果分析analyzer = QueryResultAnalyzer()if "basic" in indexes:sample_query = "公司的核心业务有哪些？"response = indexes["basic"].as_query_engine().query(sample_query)analyzer.analyze_response(response, sample_query)if __name__ == "__main__":# 假设已经构建了索引openai_api_key = os.getenv("OPENAI_API_KEY")if openai_api_key:# 加载文档documents = run_document_loader_demos()# 构建索引indexes = run_indexing_demos(documents, openai_api_key)# 运行查询演示run_query_engine_demos(indexes, openai_api_key)

自定义索引结构与高级功能

知识图谱索引

from llama_index import KnowledgeGraphIndex
from llama_index.graph_stores import SimpleGraphStore
from llama_index.llms import OpenAI
import networkx as nx
import matplotlib.pyplot as pltclass KnowledgeGraphDemo:"""知识图谱索引演示"""def __init__(self, openai_api_key: str):self.llm = OpenAI(api_key=openai_api_key)self.graph_store = SimpleGraphStore()def build_knowledge_graph(self, documents):"""构建知识图谱索引"""print("=== 构建知识图谱索引 ===")# 创建知识图谱索引kg_index = KnowledgeGraphIndex.from_documents(documents,graph_store=self.graph_store,llm=self.llm,max_triplets_per_chunk=5,include_embeddings=True)print("知识图谱构建完成")return kg_indexdef query_knowledge_graph(self, kg_index, query: str):"""查询知识图谱"""print(f"\n知识图谱查询: {query}")# 创建知识图谱查询引擎query_engine = kg_index.as_query_engine(include_text=True,response_mode="tree_summarize")response = query_engine.query(query)return responsedef visualize_graph(self, kg_index):"""可视化知识图谱"""try:# 获取图数据graph = self.graph_store.get()# 创建NetworkX图G = nx.DiGraph()# 添加节点和边for triplet in graph:subject, relation, obj = tripletG.add_node(subject, type="entity")G.add_node(obj, type="entity")G.add_edge(subject, obj, label=relation)# 绘制图形plt.figure(figsize=(12, 8))pos = nx.spring_layout(G, k=1, iterations=50)# 绘制节点nx.draw_networkx_nodes(G, pos, node_size=500, node_color='lightblue')# 绘制边nx.draw_networkx_edges(G, pos, edge_color='gray', arrows=True)# 绘制标签nx.draw_networkx_labels(G, pos, font_size=8)edge_labels = nx.get_edge_attributes(G, 'label')nx.draw_networkx_edge_labels(G, pos, edge_labels=edge_labels, font_size=6)plt.title("知识图谱可视化")plt.axis('off')plt.tight_layout()plt.savefig('knowledge_graph.png', dpi=300, bbox_inches='tight')plt.show()print("知识图谱已保存为 knowledge_graph.png")except Exception as e:print(f"图谱可视化失败: {e}")# 运行知识图谱演示
def run_knowledge_graph_demo(documents, openai_api_key: str):"""运行知识图谱演示"""kg_demo = KnowledgeGraphDemo(openai_api_key)kg_index = kg_demo.build_knowledge_graph(documents)# 测试查询test_queries = ["公司有哪些业务和技术之间的关系？","提取文档中的所有实体和关系"]for query in test_queries:response = kg_demo.query_knowledge_graph(kg_index, query)print(f"查询: {query}")print(f"回复: {response}")print("=" * 60)# 可视化知识图谱kg_demo.visualize_graph(kg_index)return kg_index

企业级文档问答系统

from typing import List, Dict, Any
import logging
from datetime import datetimeclass EnterpriseDocumentQA:"""企业级文档问答系统"""def __init__(self, openai_api_key: str, persist_dir: str = "./vector_store"):"""初始化企业级文档问答系统Args:openai_api_key: OpenAI API密钥persist_dir: 向量存储持久化目录"""self.openai_api_key = openai_api_keyself.persist_dir = Path(persist_dir)self.persist_dir.mkdir(exist_ok=True)# 设置日志self.setup_logging()# 初始化组件self.setup_components()# 加载或创建索引self.index = self.load_or_create_index()def setup_logging(self):"""设置日志系统"""logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',handlers=[logging.FileHandler('document_qa.log'),logging.StreamHandler()])self.logger = logging.getLogger(__name__)def setup_components(self):"""初始化组件"""self.llm = OpenAI(api_key=self.openai_api_key, temperature=0.1)self.embed_model = OpenAIEmbedding(api_key=self.openai_api_key)self.service_context = ServiceContext.from_defaults(llm=self.llm,embed_model=self.embed_model,chunk_size=1024,chunk_overlap=100)def load_or_create_index(self):"""加载或创建索引"""index_path = self.persist_dir / "index.json"if index_path.exists():self.logger.info("加载现有索引...")# 这里应该实现索引加载逻辑return Noneelse:self.logger.info("创建新索引...")return Nonedef ingest_documents(self, document_paths: List[str]):"""摄入文档到系统"""self.logger.info(f"开始摄入 {len(document_paths)} 个文档")all_documents = []for path in document_paths:try:if path.endswith('.pdf'):documents = self.load_pdf_documents(path)elif path.startswith('http'):documents = self.load_web_documents([path])else:documents = SimpleDirectoryReader(input_files=[path]).load_data()all_documents.extend(documents)self.logger.info(f"成功摄入文档: {path}")except Exception as e:self.logger.error(f"文档摄入失败 {path}: {e}")# 构建索引self.index = VectorStoreIndex.from_documents(all_documents,service_context=self.service_context)# 持久化索引self.index.storage_context.persist(persist_dir=str(self.persist_dir))self.logger.info(f"索引构建完成，共处理 {len(all_documents)} 个文档")return len(all_documents)def query(self, question: str, **kwargs) -> Dict[str, Any]:"""执行查询"""start_time = datetime.now()try:# 创建查询引擎query_engine = self.index.as_query_engine(similarity_top_k=kwargs.get('similarity_top_k', 5),response_mode=kwargs.get('response_mode', 'compact'))# 执行查询response = query_engine.query(question)# 计算响应时间response_time = (datetime.now() - start_time).total_seconds()# 构建响应结果result = {'question': question,'answer': str(response),'response_time': response_time,'sources': [],'success': True}# 提取源文档信息if hasattr(response, 'source_nodes'):for node in response.source_nodes[:3]:  # 最多显示3个源source_info = {'content_preview': node.node.text[:200] + '...','metadata': getattr(node.node, 'metadata', {}),'score': getattr(node, 'score', None)}result['sources'].append(source_info)self.logger.info(f"查询成功: {question} (耗时: {response_time:.2f}s)")except Exception as e:self.logger.error(f"查询失败: {question} - {e}")result = {'question': question,'answer': f"查询失败: {str(e)}",'response_time': (datetime.now() - start_time).total_seconds(),'sources': [],'success': False}return resultdef batch_query(self, questions: List[str]) -> List[Dict[str, Any]]:"""批量查询"""results = []for question in questions:result = self.query(question)results.append(result)return resultsdef get_system_stats(self) -> Dict[str, Any]:"""获取系统统计信息"""if hasattr(self.index, 'index_struct'):node_count = len(self.index.index_struct.nodes_dict)else:node_count = 0return {'index_type': type(self.index).__name__,'node_count': node_count,'persist_dir': str(self.persist_dir),'llm_model': self.llm.model}# 使用示例
def demo_enterprise_qa_system():"""演示企业级问答系统"""openai_api_key = os.getenv("OPENAI_API_KEY")if not openai_api_key:print("请设置 OPENAI_API_KEY 环境变量")return# 创建问答系统实例qa_system = EnterpriseDocumentQA(openai_api_key)# 摄入示例文档sample_docs = ["公司介绍.md","产品服务.md", "技术文档.md"]# 创建示例文档for doc_path in sample_docs:with open(doc_path, 'w', encoding='utf-8') as f:if doc_path == "公司介绍.md":f.write("""公司成立于2020年，专注于人工智能技术研发。
主要业务包括智能客服、数据分析和企业数字化转型。
我们的使命是用AI技术推动行业创新。""")elif doc_path == "产品服务.md":f.write("""智能客服系统：基于大语言模型的对话系统。
数据分析平台：提供数据洞察和预测分析。
定制开发服务：根据企业需求定制AI解决方案。""")elif doc_path == "技术文档.md":f.write("""技术架构：微服务架构，容器化部署。
技术栈：Python, FastAPI, React, PostgreSQL, Redis。
开发流程：敏捷开发，持续集成。""")# 摄入文档doc_count = qa_system.ingest_documents(sample_docs)print(f"成功摄入 {doc_count} 个文档")# 测试查询test_questions = ["公司的主要业务是什么？","技术架构采用什么方案？","有哪些产品和服务？","公司的使命是什么？"]print("\n=== 问答系统测试 ===")for question in test_questions:result = qa_system.query(question)print(f"\n问题: {question}")print(f"回答: {result['answer']}")print(f"响应时间: {result['response_time']:.2f}s")print(f"源文档数量: {len(result['sources'])}")# 显示系统统计stats = qa_system.get_system_stats()print(f"\n=== 系统统计 ===")for key, value in stats.items():print(f"{key}: {value}")if __name__ == "__main__":demo_enterprise_qa_system()