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

SDK：Software Development Kit，它是一组软件工具和资源的集合，旨在帮助开发者创建、测试、部署和维护应用程序或软件。

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

大语言模型开发框架的价值，是让开发者可以更方便地开发基于大语言模型的应用。主要提供两类帮助：

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

好的开发框架，需要具备以下特点：

1. 可靠性、鲁棒性高
2. 可维护性高
3. 可扩展性高
4. 学习成本低

举些通俗的例子：

• 与外部功能解依赖
  • 比如可以随意更换 LLM 而不用大量重构代码
  • 更换三方工具也同理
• 经常变的部分要在外部维护而不是放在代码里
  • 比如 Prompt 模板
• 各种环境下都适用
  • 比如线程安全
• 方便调试和测试
  • 至少要能感觉到用了比不用方便吧
  • 合法的输入不会引发框架内部的报错

举例：使用4行代码实现一个 RAG 系统。

注：LlamaIndex 默认的 Embedding 模型是 OpenAIEmbedding(model="text-embedding-ada-002")

环境准备

pip install --upgrade llama-index
pip install llama-index-llms-dashscope
pip install llama-index-llms-openai-like
pip install llama-index-embeddings-dashscope

rag 系统实现 demo：

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

# LlamaIndex默认使用的大模型被替换为百炼
# Settings.llm = OpenAILike(
#     model="qwen-max",
#     api_base="https://dashscope.aliyuncs.com/compatible-mode/v1",
#     api_key=os.getenv("DASHSCOPE_API_KEY"),
#     is_chat_model=True
# )

Settings.llm = DashScope(model_name=DashScopeGenerationModels.QWEN_MAX, api_key=os.getenv("DASHSCOPE_API_KEY"))

# LlamaIndex默认使用的Embedding模型被替换为百炼的Embedding模型
Settings.embed_model = DashScopeEmbedding(
    # model_name="text-embedding-v1"
    model_name=DashScopeTextEmbeddingModels.TEXT_EMBEDDING_V1,
    # api_key=os.getenv("DASHSCOPE_API_KEY")
)

# 前面的内容是准备 embedding
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader

documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()
response = query_engine.query("deepseek v3有多少参数？")

print(response)

2. LlamaIndex 介绍

官网标题：「 Build AI Knowledge Assistants over your enterprise data 」

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

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/

LlamaIndex 是一个开源框架，Github 链接：https://github.com/run-llama

LlamaIndex 的核心模块

安装 Llamaindex

pip install llama-index

3. 数据加载

3.1. 加载本地数据

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

3.1.1. 使用 SimpleDirectoryReader 加载文本数据

定义基础函数:

import json
from pydantic.v1 import BaseModel

def 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")
    

使用 SimpleDirectoryReader 加载文本，不指定文本解析器时，默认使用的 PDFReader。

from llama_index.core import SimpleDirectoryReader

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

默认的 PDFReader 解析效果不理想，SimpleDirectoryReader 可以指定文本解析器，我们来使用LlamaIndex 官方的文本解析器 LlamaParse。要想使用该解析器，需要先安装llamaIndex云服务依赖：

pip install llama-cloud-services

使用时还需要申请账号，有一定的免费额度。

# 在系统环境变量里配置 LLAMA_CLOUD_API_KEY=XXX

from llama_cloud_services import LlamaParse
from llama_index.core import SimpleDirectoryReader
import nest_asyncio
nest_asyncio.apply() # 只在Jupyter笔记环境中需要此操作，否则会报错

# set up parser
parser = 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)

3.2. Data Connectors

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

pip install llama-index-readers-web

from llama_index.readers.web import SimpleWebPageReader

documents = SimpleWebPageReader(html_to_text=True).load_data(
    ["https://edu.guangjuke.com/tx/"]
)

print(documents[0].text)

更多的 Data Connectors 参考 LlamaIndex 文档，如连接第三方服务的数据加载器，内置的文件加载器，更多加载器可以在 LlamaHub 上找到。

4. 文本切分与解析（Chunking）

为方便检索，我们通常把 Document 切分为 Node。在 LlamaIndex 中，Node 被定义为一个文本的 chunk。

4.1. 使用 TextSplitters 对文本做切分

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

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

例如：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
)

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

NodeParser 包括 MarkdownNodeParser，JSONNodeParser、HTMLNodeParser 等。例如：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://developers.llamaindex.ai/"]
)

# 默认解析 ["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")

5. 索引（Indexing）与检索（Retrieval）

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

5.1. 向量检索

1. 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数学能力怎么样？")

print(results[0].text)

2. 使用自定义的 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数学能力怎么样")

print(results[0])

5.2. 更多索引与检索方式

LlamaIndex 内置了丰富的检索机制，例如：

• 关键字检索

  • BM25Retriever：基于 tokenizer 实现的 BM25 经典检索算法
  • KeywordTableGPTRetriever：使用 GPT 提取检索关键字
  • KeywordTableSimpleRetriever：使用正则表达式提取检索关键字
  • KeywordTableRAKERetriever：使用RAKE算法提取检索关键字（有语言限制）

• RAG-Fusion QueryFusionRetriever

• 还支持 KnowledgeGraph、SQL、Text-to-SQL 等等

5.3. 检索后处理

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

6. 生成回复（QA & Chat）

6.1 单轮回复

普通输出：

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

6.2 多轮对话（Chat Engine）

多轮对话使用 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)

7. 底层接口：Prompt、LLM 与 Embedding

7.1 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="这是什么"
    )
)

7.2 语言模型

使用 OpenAI 大模型：

from llama_index.llms.openai import OpenAI

llm = 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)

使用 DeepSeek：

# !pip install llama-index-llms-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)

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

7.3 Embedding 模型

之前我们介绍过阿里百炼的 embeddings 模型，这里我们看下 Open AI 的 embeddings 模型。

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

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

8. 基于 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}")