Llamaindex框架

大语言模型开发框架的价值#

所有开发框架（SDK）的核心价值，都是降低开发、维护成本。

大语言模型开发框架的价值，是让开发者可以更方便地开发基于大语言模型的应用：

第三方能力抽象。比如 LLM、向量数据库、搜索接口等
常用工具、方案封装
底层实现封装。比如流式接口、超时重连、异步与并行等

使用Llamaindex几行代码实现一个简易的RAG

import os
from llama_index.core import Settings
from llama_index.llms.openai_like import OpenAILike
from llama_index.llms.dashscope import DashScope, DashScopeGenerationModels
from llama_index.embeddings.dashscope import DashScopeEmbedding, DashScopeTextEmbeddingModels


# 设置全局的框架模型
Settings.llm = DashScope(model_name=DashScopeGenerationModels.QWEN_MAX, api_key=os.getenv("DASHSCOPE_API_KEY"))

# 设置LlamaIndex默认使用的Embedding模型
# 如果不设置，默认的embedding是OpenAIEmbedding(model="text-embedding-ada-002")
Settings.embed_model = DashScopeEmbedding(
    model_name=DashScopeTextEmbeddingModels.TEXT_EMBEDDING_V1
   #设置api_key如果不显示设置，也会自动执行下面代码从环境变量获取，必须是DASHSCOPE_API_KEY
    api_key=os.getenv("DASHSCOPE_API_KEY") 
)

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader

# 读取./data目录下的所有文件内容
documents = SimpleDirectoryReader("./data").load_data()
# 把文档变成向量数据库
index = VectorStoreIndex.from_documents(documents)
# 根据向量库创建问答机器人
query_engine = index.as_query_engine()
# 提问生成回答
response = query_engine.query("DEEPSEEK的发展历程")
print(response)

LlamaIndex框架#

框架简介#

LlamaIndex 是一个为开发「知识增强」的大语言模型应用的开源框架。知识增强泛指任何在私有或特定领域数据基础上应用大语言模型的情况

Question-Answering Chatbots (也就是 RAG)
Document Understanding and Extraction （文档理解与信息抽取）
Autonomous Agents that can perform research and take actions （智能体应用）
Workflow orchestrating single and multi-agent (编排单个或多个智能体形成工作流）

LlamaIndex 有 Python 和 Typescript 两个版本，Python 版的文档相对更完善。

Python 文档地址：https://docs.llamaindex.ai/en/stable/
Python API 接口文档：https://docs.llamaindex.ai/en/stable/api_reference/
TS 文档地址：https://ts.llamaindex.ai/
Github 链接：https://github.com/run-llama

#安装Llamaindex
pip install llama-index

核心模块#

数据加载(Loading)#

加载本地数据#

SimpleDirectoryReader 是一个简单的本地文件加载器。它会遍历指定目录，并根据文件扩展名自动加载文件（文本内容）。

支持的文件类型：

.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

from llama_index.core import SimpleDirectoryReader

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

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

对图像、视频、语音类文件，默认不会自动提取其中文字。如需提取，参考后面的Data Connectors

更换文件加载器#

SimpleDirectoryReader默认的PDFReader效果并不理想，可以更换文件加载器

LlamaParse

云端商业API 服务
从 https://cloud.llamaindex.ai 注册并获取 api-key
```
pip install llama-cloud-servicestm 
```
效果比PDFReader强，但是收费以及实际效果不如minerU，具体取决于使用场景，推荐使用minerU

# 示例
from llama_cloud_services import LlamaParse
from llama_index.core import SimpleDirectoryReader
import nest_asyncio

# nest_asyncio.apply() 如果是Jupyter环境中需要此操作，否则会报错


parser = LlamaParse(
    result_type="markdown"  # 将文件输出成md
)
file_extractor = {".pdf": parser}

documents = SimpleDirectoryReader(input_dir="./data", required_exts=[".pdf"], file_extractor=file_extractor).load_data()
print(documents[0].text)

MinerU

官网：https://mineru.net
Github:https://github.com/opendatalab/mineru
支持在线使用、下载客户端、以及API
可以本地化部署

Data Connectors#

用于处理更丰富的数据类型，并将其读取为 Document 的形式，例如直接读取网页文档

from llama_index.readers.web import SimpleWebPageReader

documents = SimpleWebPageReader(html_to_text=True).load_data(
    ["https://www.sina.com.cn"]
)

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/
Llamhub:https://llamahub.ai/

文本切分与解析(Chunking)#

为方便检索，我们通常把 Document 切分为 Node。

在 LlamaIndex 中，Node 被定义为一个文本的「chunk」。

使用TextSplitters对文本做切分#

# TokenTextSplitter按指定 token 数切分文本
from llama_index.core import Document
from llama_index.core.node_parser import TokenTextSplitter

node_parser = TokenTextSplitter(
    chunk_size=512,  # 每个 chunk 的最大长度
    chunk_overlap=200  # chunk 之间重叠长度
)

nodes = node_parser.get_nodes_from_documents(
    documents, show_progress=False
)

LlamaIndex 提供了丰富的 TextSplitter，例如：

SentenceSplitter：在切分指定长度的 chunk 同时尽量保证句子边界不被切断；
CodeSplitter：根据 AST（编译器的抽象句法树）切分代码，保证代码功能片段完整；
SemanticSplitterNodeParser：根据语义相关性对将文本切分为片段。

使用 NodeParsers 对有结构的文档做解析#

# HTMLNodeParser解析 HTML 文档
from llama_index.core.node_parser import HTMLNodeParser
from llama_index.readers.web import SimpleWebPageReader

documents = SimpleWebPageReader(html_to_text=False).load_data(
    ["https://www.sina.com.cn"]
)

# 默认解析 ["p", "h1", "h2", "h3", "h4", "h5", "h6", "li", "b", "i", "u", "section"]
parser = HTMLNodeParser(tags=["span"])  # 可以自定义解析哪些标签
nodes = parser.get_nodes_from_documents(documents)

for node in nodes:
    print(node.text+"\n")

更多的 NodeParser 包括 MarkdownNodeParser，JSONNodeParser等等

索引(indexing)与检索(Retrieval)#

基础概念：在「检索」相关的上下文中，「索引」即index，通常是指为了实现快速检索而设计的特定「数据结构」。

索引的具体原理与实现可以参考可以参考：传统索引、向量索引

向量检索#

VectorStoreIndex 直接在内存中构建一个 Vector Store 并建索引

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.node_parser import TokenTextSplitter, SentenceSplitter

# 加载 pdf 文档
documents = SimpleDirectoryReader(
    "./data", 
    required_exts=[".pdf"],
).load_data()

# 定义 Node Parser
node_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")

# 获取 retriever
vector_retriever = index.as_retriever(
    similarity_top_k=2 # 返回2个结果
)

# 检索
results = vector_retriever.retrieve("deepseek v3和v2对比？")

print(results[0].text)

使用自定义的 Vector Store，以 Qdrant 为例：

pip install llama-index-vector-stores-qdrant

from llama_index.core.indices.vector_store.base import VectorStoreIndex
from llama_index.vector_stores.qdrant import QdrantVectorStore
from llama_index.core import StorageContext

from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Distance

client = 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 Store
index = VectorStoreIndex(nodes, storage_context=storage_context)

# 获取 retriever
vector_retriever = index.as_retriever(similarity_top_k=1)

# 检索
results = vector_retriever.retrieve("deepseek v3和v2对比？")

print(results[0])

检索后处理#

LlamaIndex 的 Node Postprocessors 提供了一系列检索后处理模块。

例如：我们可以用不同模型对检索后的 Nodes 做重排序

# 获取 retriever
vector_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 LLMRerank

postprocessor = LLMRerank(top_n=2)

nodes = postprocessor.postprocess_nodes(nodes, query_str="deepseek v3有多少参数?")

for i, node in enumerate(nodes):
    print(f"[{i}] {node.text}")

更多的 Rerank 及其它后处理方法，参考官方文档：Node Postprocessor Modules

生成回复(QA & CHAT)#

单轮回答#

不会基于上下文，只是基于本次的问答

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)#

多轮对话会基于上下文

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)

底层接口#

Prompt模版#

PromptTemplate 单轮提示词模板

from llama_index.core import PromptTemplate

prompt = PromptTemplate("写一篇关于{topic}的笔记")

prompt.format(topic="小红")

ChatPromptTemplate多轮提示词模版

from llama_index.core.llms import ChatMessage, MessageRole
from llama_index.core import ChatPromptTemplate

chat_text_qa_msgs = [
    ChatMessage(
        role=MessageRole.SYSTEM,
        content="你叫{name}，你必须根据用户提供的上下文回答问题。",
    ),
    ChatMessage(
        role=MessageRole.USER, 
        content=(
            "已知上下文：\n" \
            "{context}\n\n" \
            "问题：{question}"
        )
    ),
]
text_qa_template = ChatPromptTemplate(chat_text_qa_msgs)

print(
    text_qa_template.format(
        name="小明",
        context="这是一个测试",
        question="这是什么"
    )
)

LLM语言模型#

LlamaIndex 已集成多个大语言模型，包括云服务 API 和本地部署 API，详见官方文档：Available LLM integrations

# openai
from llama_index.llms.openai import OpenAI

llm = OpenAI(temperature=0, model="gpt-4o")
response = llm.complete(prompt.format(topic="小明"))

print(response.text)

# deepseek
import os
from llama_index.llms.deepseek import DeepSeek

llm = DeepSeek(model="deepseek-chat", api_key=os.getenv("DEEPSEEK_API_KEY"), temperature=1.5)

response = llm.complete("写个笑话")
print(response)

# 设置全局使用的语言模型
from llama_index.core import Settings

Settings.llm = DeepSeek(model="deepseek-chat", api_key=os.getenv("DEEPSEEK_API_KEY"), temperature=1.5)

Embedding#

LlamaIndex 同样集成了多种 Embedding 模型，包括云服务 API 和开源模型（HuggingFace）等，详见官方文档。

from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core import Settings

# 全局设定
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small", dimensions=512)

基于LlamaIndex实现功能较完整的RAG#

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

# 向量库
from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Distance

EMBEDDING_DIM = 1536
COLLECTION_NAME = "full_demo"
PATH = "./qdrant_db"

client = QdrantClient(path=PATH)

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, get_response_synthesizer
from llama_index.vector_stores.qdrant import QdrantVectorStore
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.response_synthesizers import ResponseMode
from llama_index.core.ingestion import IngestionPipeline
from llama_index.core import Settings
from llama_index.core import StorageContext
from llama_index.core.postprocessor import LLMRerank, SimilarityPostprocessor
from llama_index.core.retrievers import QueryFusionRetriever
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.chat_engine import CondenseQuestionChatEngine
from llama_index.llms.dashscope import DashScope, DashScopeGenerationModels
from 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 Pipeline
Settings.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. 创建 collection
client.create_collection(
    collection_name=COLLECTION_NAME,
    vectors_config=VectorParams(size=EMBEDDING_DIM, distance=Distance.COSINE)
)

# 5. 创建 Vector Store
vector_store = QdrantVectorStore(client=client, collection_name=COLLECTION_NAME)

# 6. 指定 Vector Store 的 Storage 用于 index
storage_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 engine
query_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}")

作者：木子七

出处：https://www.cnblogs.com/Mickey-7/p/20016143

相关阅读:
每天学习2小时——黑客（网络安全）技术
 42.接雨水--dp版本
 Win2016安装安全狗和DVWA
1.4.14 实验14：ospf多区域
 MTK cts测试注意事项
 Mysql集群及高可用-并行复制5
特征选择-sklearn
Jmeter状态码及请求
 云计算 - 4 - Spark的安装与应用
 整形数据和浮点型数据在内存中的存储差别
原文地址：https://www.cnblogs.com/Mickey-7/p/20016143