LlamaIndex知识管理与信息检索

- Python 文档地址：https://docs.llamaindex.ai/en/stable/-Python API文档：https://docs.llamaindex.ai/en/stable/api_reference/-TS文档地址：https://ts.llamaindex.ai/LlamaIndex 是一个开源框架Github 链接：https://github.com/run-llama

pip install llama-index

支持的文件类型：- `.csv` - comma-separated values- `.docx` - Microsoft Word- `.epub` - EPUB ebook format- `.hwp` - Hangul Word Processor- `.ipynb` - Jupyter Notebook- `.jpeg`, `.jpg` - JPEG image- `.mbox` - MBOX email archive- `.md` - Markdown- `.mp3`, `.mp4` - audio and video- `.pdf` - Portable Document Format- `.png` - Portable Network Graphics- `.ppt`, `.pptm`, `.pptx` - Microsoft PowerPoint

import jsonfrom pydantic.v1 import BaseModeldef show_json(data):    """用于展示json数据"""    if isinstance(data, str):        obj = json.loads(data)        print(json.dumps(obj, indent=4, ensure_ascii=False))    elif isinstance(data, dict) or isinstance(data, list):        print(json.dumps(data, indent=4, ensure_ascii=False))    elif issubclass(type(data), BaseModel):        print(json.dumps(data.dict(), indent=4, ensure_ascii=False))def show_list_obj(data):    """用于展示一组对象"""    if isinstance(data, list):        for item in data:            show_json(item)    else:        raise ValueError("Input is not a list")

from llama_index.core import SimpleDirectoryReaderreader = SimpleDirectoryReader(        input_dir="./data", # 目标目录        recursive=False, # 是否递归遍历子目录        required_exts=[".pdf"] # (可选)只读取指定后缀的文件    )documents = reader.load_data()

print(documents[0].text)show_json(documents[0].json())

pip install llama-cloud-services

# 在系统环境变量里配置 LLAMA_CLOUD_API_KEY=XXXfrom llama_cloud_services import LlamaParsefrom llama_index.core import SimpleDirectoryReaderimport nest_asyncionest_asyncio.apply() # 只在Jupyter笔记环境中需要此操作，否则会报错# set up parserparser = LlamaParse(    result_type="markdown"  # "markdown" and "text" are available)file_extractor = {".pdf": parser}documents = SimpleDirectoryReader(input_dir="./data", required_exts=[".pdf"], file_extractor=file_extractor).load_data()print(documents[0].text)

pip install llama-index-readers-web
from llama_index.readers.web import SimpleWebPageReaderdocuments = SimpleWebPageReader(html_to_text=True).load_data(    ["https://www.baidu.com/"])print(documents[0].text)

更多 Data Connectors内置的文件加载器:https://llamahub.ai/l/readers/llama-index-readers-file连接三方服务的数据加载器:https://docs.llamaindex.ai/en/stable/module_guides/loading/connector/modules/更多加载器可以在LlamaHub:https://llamahub.ai/ 上找到

例如：`TokenTextSplitter` 按指定 token 数切分文本from  llama_index.core import Documentfrom llama_index.core.node_parser import TokenTextSplitternode_parser = TokenTextSplitter( chunk_size=512,  # 每个 chunk 的最大长度     chunk_overlap=200 # chunk 之间重叠长度)nodes = node_parser.get_nodes_from_documents(    documents,     show_progress=False)show_json(nodes[1].json())show_json(nodes[2].json())

from llama_index.core import VectorStoreIndex, SimpleDirectoryReaderfrom llama_index.core.node_parser import TokenTextSplitter, SentenceSplitter
# 加载 pdf 文档documents = SimpleDirectoryReader(    "./data",     required_exts=[".pdf"],).load_data()
# 定义 Node Parsernode_parser = TokenTextSplitter(chunk_size=512, chunk_overlap=200)
# 切分文档nodes = node_parser.get_nodes_from_documents(documents)
# 构建 index，默认是在内存中index = VectorStoreIndex(nodes)
# 另外一种实现方式# index = VectorStoreIndex.from_documents(documents=documents, transformations=[SentenceSplitter(chunk_size=512)])
# 写入本地文件# index.storage_context.persist(persist_dir="./doc_emb")
# 获取 retrievervector_retriever = index.as_retriever(    similarity_top_k=2 # 返回2个结果)
# 检索results = vector_retriever.retrieve("DeepSeek v3数学能力怎么样？")
print(results[0].text)

pip install llama-index-vector-stores-qdrant

from llama_index.core.indices.vector_store.base import VectorStoreIndexfrom llama_index.vector_stores.qdrant import QdrantVectorStorefrom llama_index.core import StorageContextfrom qdrant_client import QdrantClientfrom qdrant_client.models import VectorParams, Distanceclient = QdrantClient(location=":memory:")collection_name = "demo"collection = client.create_collection(    collection_name=collection_name,    vectors_config=VectorParams(size=1536, distance=Distance.COSINE))vector_store = QdrantVectorStore(client=client, collection_name=collection_name)# storage: 指定存储空间storage_context = StorageContext.from_defaults(vector_store=vector_store)# 创建 index：通过 Storage Context 关联到自定义的 Vector Storeindex = VectorStoreIndex(nodes, storage_context=storage_context)# 获取 retrievervector_retriever = index.as_retriever(similarity_top_k=1)# 检索results = vector_retriever.retrieve("deepseek v3数学能力怎么样")print(results[0])### 5.2、更多索引与检索方式

LlamaIndex 内置了丰富的检索机制，例如：- 关键字检索  - [`BM25Retriever`](https://docs.llamaindex.ai/en/stable/api_reference/retrievers/bm25/)：基于 tokenizer 实现的 BM25 经典检索算法  - [`KeywordTableGPTRetriever`](https://docs.llamaindex.ai/en/stable/api_reference/retrievers/keyword/#llama_index.core.indices.keyword_table.retrievers.KeywordTableGPTRetriever)：使用 GPT 提取检索关键字  - [`KeywordTableSimpleRetriever`](https://docs.llamaindex.ai/en/stable/api_reference/retrievers/keyword/#llama_index.core.indices.keyword_table.retrievers.KeywordTableSimpleRetriever)：使用正则表达式提取检索关键字  - [`KeywordTableRAKERetriever`](https://docs.llamaindex.ai/en/stable/api_reference/retrievers/keyword/#llama_index.core.indices.keyword_table.retrievers.KeywordTableRAKERetriever)：        使用[`RAKE`](https://pypi.org/project/rake-nltk/)算法提取检索关键字（有语言限制）- RAG-Fusion [`QueryFusionRetriever`](https://docs.llamaindex.ai/en/stable/api_reference/retrievers/query_fusion/)- 还支持 [KnowledgeGraph](https://docs.llamaindex.ai/en/stable/api_reference/retrievers/knowledge_graph/)[SQL](https://docs.llamaindex.ai/en/stable/api_reference/retrievers/sql/#llama_index.core.retrievers.SQLRetriever)[Text-to-SQL](https://docs.llamaindex.ai/en/stable/api_reference/retrievers/sql/#llama_index.core.retrievers.NLSQLRetriever) 等等

# 获取 retrievervector_retriever = index.as_retriever(similarity_top_k=5)# 检索nodes = vector_retriever.retrieve("deepseek v3有多少参数?")for i, node in enumerate(nodes):    print(f"[{i}] {node.text}\n")

from llama_index.core.postprocessor import LLMRerankpostprocessor = LLMRerank(top_n=2)nodes = postprocessor.postprocess_nodes(nodes, query_str="deepseek v3有多少参数?")for i, node in enumerate(nodes):    print(f"[{i}] {node.text}")

qa_engine = index.as_query_engine()response = qa_engine.query("deepseek v3数学能力怎么样?")print(response)

qa_engine = index.as_query_engine(streaming=True)response = qa_engine.query("deepseek v3数学能力怎么样?")response.print_response_stream()

chat_engine = index.as_chat_engine()response = chat_engine.chat("deepseek v3数学能力怎么样?")print(response)

response = chat_engine.chat("代码能力呢?")print(response)

chat_engine = index.as_chat_engine()streaming_response = chat_engine.stream_chat("deepseek v3数学能力怎么样?")# streaming_response.print_response_stream()for token in streaming_response.response_gen:    print(token, end="", flush=True)

from llama_index.llms.openai import OpenAIllm = OpenAI(temperature=0, model="gpt-4o")

response = llm.complete(prompt.format(topic="小明"))print(response.text)

response = llm.complete(    text_qa_template.format(        name="小明",        context="这是一个测试",        question="你是谁，我们在干嘛"    ))print(response.text)

pip install llama-index-llms-deepseek

import osfrom llama_index.llms.deepseek import DeepSeekllm = DeepSeek(model="deepseek-chat", api_key=os.getenv("DEEPSEEK_API_KEY"), temperature=1.5)response = llm.complete("作首唐诗")print(response)

from llama_index.core import SettingsSettings.llm = DeepSeek(model="deepseek-chat", api_key=os.getenv("DEEPSEEK_API_KEY"), temperature=1.5)

from llama_index.embeddings.openai import OpenAIEmbeddingfrom llama_index.core import Settings# 全局设定Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small", dimensions=512)

功能要求：
- 加载指定目录的文件- 支持 RAG-Fusion- 使用 Qdrant 向量数据库，并持久化到本地- 支持检索后排序- 支持多轮对话

from qdrant_client import QdrantClientfrom qdrant_client.models import VectorParams, DistanceEMBEDDING_DIM = 1536COLLECTION_NAME = "full_demo"PATH = "./qdrant_db"client = QdrantClient(path=PATH)

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, get_response_synthesizerfrom llama_index.vector_stores.qdrant import QdrantVectorStorefrom llama_index.core.node_parser import SentenceSplitterfrom llama_index.core.response_synthesizers import ResponseModefrom llama_index.core.ingestion import IngestionPipelinefrom llama_index.core import Settingsfrom llama_index.core import StorageContextfrom llama_index.core.postprocessor import LLMRerank, SimilarityPostprocessorfrom llama_index.core.retrievers import QueryFusionRetrieverfrom llama_index.core.query_engine import RetrieverQueryEnginefrom llama_index.core.chat_engine import CondenseQuestionChatEnginefrom llama_index.llms.dashscope import DashScope, DashScopeGenerationModelsfrom llama_index.embeddings.dashscope import DashScopeEmbedding, DashScopeTextEmbeddingModels# 1. 指定全局llm与embedding模型Settings.llm = DashScope(model_name=DashScopeGenerationModels.QWEN_MAX,api_key=os.getenv("DASHSCOPE_API_KEY"))Settings.embed_model = DashScopeEmbedding(model_name=DashScopeTextEmbeddingModels.TEXT_EMBEDDING_V1)# 2. 指定全局文档处理的 Ingestion PipelineSettings.transformations = [SentenceSplitter(chunk_size=512, chunk_overlap=200)]# 3. 加载本地文档documents = SimpleDirectoryReader("./data").load_data()if client.collection_exists(collection_name=COLLECTION_NAME):    client.delete_collection(collection_name=COLLECTION_NAME)# 4. 创建 collectionclient.create_collection(    collection_name=COLLECTION_NAME,    vectors_config=VectorParams(size=EMBEDDING_DIM, distance=Distance.COSINE))# 5. 创建 Vector Storevector_store = QdrantVectorStore(client=client, collection_name=COLLECTION_NAME)# 6. 指定 Vector Store 的 Storage 用于 indexstorage_context = StorageContext.from_defaults(vector_store=vector_store)index = VectorStoreIndex.from_documents(    documents, storage_context=storage_context)# 7. 定义检索后排序模型reranker = LLMRerank(top_n=2)# 最终打分低于0.6的文档被过滤掉sp = SimilarityPostprocessor(similarity_cutoff=0.6)# 8. 定义 RAG Fusion 检索器fusion_retriever = QueryFusionRetriever(    [index.as_retriever()],    similarity_top_k=5, # 检索召回 top k 结果    num_queries=3,  # 生成 query 数    use_async=False,    # query_gen_prompt="",  # 可以自定义 query 生成的 prompt 模板)# 9. 构建单轮 query enginequery_engine = RetrieverQueryEngine.from_args(    fusion_retriever,    node_postprocessors=[reranker],    response_synthesizer=get_response_synthesizer(        response_mode = ResponseMode.REFINE    ))# 10. 对话引擎chat_engine = CondenseQuestionChatEngine.from_defaults(    query_engine=query_engine,     # condense_question_prompt="" # 可以自定义 chat message prompt 模板)

# 测试多轮对话# User: deepseek v3有多少参数# User: 每次激活多少while True:    question=input("User:")    if question.strip() == "":        break    response = chat_engine.chat(question)    print(f"AI: {response}")