[TOC] ## 示例 import ``` pip install llama-index pip install llama-cpp-python pip install chromadb ``` 示例 ``` from llama_index import VectorStoreIndex, SimpleDirectoryReader, ServiceContext from llama_index.llms import Ollama from llama_index.embeddings import HuggingFaceEmbedding # 配置 Ollama 模型和嵌入模型 def setup_service_context(): # 使用 Ollama 的 llama2 模型 llm = Ollama(model="llama2", temperature=0) # 使用开源的 sentence-transformers 作为嵌入模型 embed_model = HuggingFaceEmbedding( model_name="sentence-transformers/all-MiniLM-L6-v2" ) # 创建服务上下文 service_context = ServiceContext.from_defaults( llm=llm, embed_model=embed_model, chunk_size=512, # 文档分块大小 chunk_overlap=128 # 分块重叠部分 ) return service_context # 加载文档并创建索引 def create_index(directory_path): # 加载文档 documents = SimpleDirectoryReader( directory_path, filename_as_id=True ).load_data() # 创建索引 service_context = setup_service_context() index = VectorStoreIndex.from_documents( documents, service_context=service_context ) return index # 问答功能 def ask_question(index, question): # 创建查询引擎 query_engine = index.as_query_engine( response_mode="compact", similarity_top_k=3 # 检索最相关的3个片段 ) # 执行查询 response = query_engine.query(question) return response # 主函数 def main(): # 指定文档目录 docs_dir = "./your_docs_folder" # 创建索引 print("正在创建文档索引...") index = create_index(docs_dir) # 交互式问答 while True: question = input("\n请输入您的问题 (输入 'q' 退出): ") if question.lower() == 'q': break try: response = ask_question(index, question) print("\n回答:", response) except Exception as e: print(f"发生错误: {e}") if __name__ == "__main__": main() ``` ## 示例:含添加持久化存储和错误处理 ``` from llama_index import StorageContext, load_index_from_storage from pathlib import Path import logging import sys class DocumentQA: def __init__(self, docs_dir, persist_dir="./storage"): self.docs_dir = docs_dir self.persist_dir = persist_dir self.setup_logging() self.index = self.initialize_index() def setup_logging(self): logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.StreamHandler(sys.stdout), logging.FileHandler('qa_system.log') ] ) self.logger = logging.getLogger(__name__) def initialize_index(self): try: # 尝试加载已存在的索引 if Path(self.persist_dir).exists(): self.logger.info("加载现有索引...") storage_context = StorageContext.from_defaults( persist_dir=self.persist_dir ) return load_index_from_storage(storage_context) # 创建新索引 self.logger.info("创建新索引...") service_context = setup_service_context() documents = SimpleDirectoryReader( self.docs_dir, filename_as_id=True ).load_data() index = VectorStoreIndex.from_documents( documents, service_context=service_context ) # 持久化存储 index.storage_context.persist(persist_dir=self.persist_dir) return index except Exception as e: self.logger.error(f"索引初始化失败: {e}") raise def refresh_index(self): """刷新文档索引""" try: self.logger.info("刷新文档索引...") # 删除旧索引 if Path(self.persist_dir).exists(): import shutil shutil.rmtree(self.persist_dir) self.index = self.initialize_index() self.logger.info("索引刷新完成") except Exception as e: self.logger.error(f"索引刷新失败: {e}") raise def ask(self, question, top_k=3): """处理问题并返回答案""" try: self.logger.info(f"处理问题: {question}") query_engine = self.index.as_query_engine( response_mode="compact", similarity_top_k=top_k ) response = query_engine.query(question) return response except Exception as e: self.logger.error(f"问答过程出错: {e}") raise # 使用示例 def run_qa_system(): try: qa_system = DocumentQA("./your_docs_folder") while True: question = input("\n请输入您的问题 (输入 'q' 退出, 'r' 刷新索引): ") if question.lower() == 'q': break elif question.lower() == 'r': qa_system.refresh_index() print("索引已刷新") continue try: response = qa_system.ask(question) print("\n回答:", response) except Exception as e: print(f"无法回答该问题: {e}") except Exception as e: print(f"系统启动失败: {e}") if __name__ == "__main__": run_qa_system() ```