大模型应用开发：从原理到实践的全面指南

文章借助AI撰写，作者已阅读，介意者勿阅！！！

2026年，大语言模型（LLM）应用开发已经形成完整的生态体系。本文从原理出发，系统讲解LLM应用开发的各个方向，帮助开发者既了解原理，又掌握实践，更能自主创新。

---

一、大模型基础：从文字到智能

在开始学习大模型应用开发之前，我们需要先理解大模型本身是如何工作的。本章将从最基础的概念开始，讲解文字是如何进入大模型、被处理、然后变成我们能看懂的输出的。

本章核心问题：

• 一段文字是如何进入大模型的？
• 大模型是如何"理解"文字的？
• 大模型是如何生成文字的？
• Embedding到底是什么？它在大模型中扮演什么角色？

1.1 大模型是如何处理文字的？

1.1.1 文字进入大模型的全过程

当你向ChatGPT发送一条消息时，背后经历了以下完整流程：

用户输入: "什么是RAG？"
        ↓
┌─────────────────────────────────────────────────────────────┐
│ 第1步：Tokenization（分词）                                  │
│ 将文字拆分成更小的单元（Token）                              │
│ "什么是RAG？" → ["什么", "是", "R", "A", "G", "？"]         │
└─────────────────────────────────────────────────────────────┘
        ↓
┌─────────────────────────────────────────────────────────────┐
│ 第2步：Token to ID（转为数字）                               │
│ 将每个Token映射到一个数字ID                                  │
│ ["什么", "是", "R", "A", "G", "？"] → [1234, 5678, 45, 32, 38, 3456] │
└─────────────────────────────────────────────────────────────┘
        ↓
┌─────────────────────────────────────────────────────────────┐
│ 第3步：Embedding（向量化）                                   │
│ 将每个Token ID转换为一个高维向量                             │
│ [1234, 5678, 45, 32, 38, 3456] → [[0.1, 0.2, ...], [0.3, 0.4, ...], ...] │
│ 每个Token变成一个4096维的向量                                │
└─────────────────────────────────────────────────────────────┘
        ↓
┌─────────────────────────────────────────────────────────────┐
│ 第4步：Transformer推理                                       │
│ 向量经过多层自注意力网络处理                                 │
│ 输入向量 → 32/64层Transformer → 输出向量                     │
└─────────────────────────────────────────────────────────────┘
        ↓
┌─────────────────────────────────────────────────────────────┐
│ 第5步：输出层（Output Layer）                                │
│ 将输出向量转换为每个Token的分数                              │
│ 输出向量 → 线性层 → Logits（50000个分数）                    │
│ 如：{"检索": 0.8, "生成": 0.1, "增强": 0.05, ...}          │
└─────────────────────────────────────────────────────────────┘
        ↓
┌─────────────────────────────────────────────────────────────┐
│ 第6步：Decoding（解码）                                      │
│ 从分数中选择一个Token，转换回文字                            │
│ 选择"检索" → "检索"                                         │
└─────────────────────────────────────────────────────────────┘
        ↓
模型输出: "RAG是检索增强生成..."

关键概念：

1.1.2 Tokenization：文字如何变成Token

Tokenization是将文本分割成更小的单元（Token）的过程。

为什么要分词？

分词示例：

# 使用tiktoken（OpenAI的分词器）
import tiktoken

encoder = tiktoken.encoding_for_model("gpt-4")

# 文本 → Token
text = "什么是RAG？"
tokens = encoder.encode(text)
print(tokens)  # [1234, 5678, 45, 32, 38, 3456]

# Token → 文本
decoded_text = encoder.decode(tokens)
print(decoded_text)  # "什么是RAG？"

# 查看每个Token是什么
for token_id in tokens:
    token_text = encoder.decode([token_id])
    print(f"ID {token_id}: '{token_text}'")
# 输出：
# ID 1234: '什么'
# ID 5678: '是'
# ID 45: 'R'
# ID 32: 'A'
# ID 38: 'G'
# ID 3456: '？'

主流分词算法详解

现代大模型主要使用以下几种分词算法：

BPE算法原理（最常用）

BPE（Byte Pair Encoding） 是当前最主流的分词算法，被GPT系列、Claude、DeepSeek等模型使用。

核心思想：从单个字符开始，迭代地合并语料中最频繁出现的相邻字符对，直到达到目标词汇表大小。

训练过程：

初始语料: "low low low low low lowest lowest newer newer newer"
初始词汇: {l, o, w, e, s, t, n, r}

第1轮: 统计所有相邻字符对
  "l o" 出现 7 次
  "o w" 出现 7 次
  "w e" 出现 2 次
  "e r" 出现 3 次
  ...
  合并频率最高的 "l o" → "lo"

第2轮: 重新统计
  "lo w" 出现 7 次
  "w e" 出现 2 次
  ...
  合并 "lo w" → "low"

第3轮: 继续合并
  "low e" 出现 2 次
  "e r" 出现 3 次
  ...
  合并 "e r" → "er"

... 重复直到达到目标词汇表大小

Byte-level BPE（GPT-2及之后的模型使用）：

• 基础词汇表为256个字节值（而非Unicode字符）
• 任何UTF-8文本都能被表示，永远不会出现未知Token
• 基础词汇表固定，不随语言变化

代码示例：

# 使用HuggingFace Tokenizers训练BPE分词器
from tokenizers import Tokenizer
from tokenizers.models import BPE
from tokenizers.trainers import BpeTrainer
from tokenizers.pre_tokenizers import ByteLevel

# 1. 初始化BPE分词器
tokenizer = Tokenizer(BPE())

# 2. 设置预分词器（Byte-level）
tokenizer.pre_tokenizer = ByteLevel(add_prefix_space=True)

# 3. 配置训练器
# vocab_size: 词汇表大小，决定分词粒度
trainer = BpeTrainer(
    vocab_size=50000,           # 词汇表大小
    min_frequency=2,            # 最小出现频率
    special_tokens=["<unk>", "<s>", "</s>", "<pad>"],
    initial_alphabet=ByteLevel.alphabet()  # 包含所有256字节
)

# 4. 训练
tokenizer.train(files=["corpus.txt"], trainer=trainer)

# 5. 使用
result = tokenizer.encode("Hello, 你好世界！")
print(f"Tokens: {result.tokens}")  # ['Hello', ',', 'Ġä½ł', '好', 'ä¸ç', 'ķ', '¼', '!']
print(f"IDs: {result.ids}")

不同模型使用的分词算法

词汇表大小趋势：

2018-2019: 30K-50K
  BERT: 30,522
  GPT-2: 50,257
  T5: 32,128

2020-2022: 50K-100K
  GPT-3: 50,257
  GPT-3.5: 100,277

2023-2024: 100K-130K
  GPT-4: 100,277
  Llama 3: 128,256
  Qwen 2: 151,643

2025-2026: 200K+
  GPT-4o: ~200,000
  Llama 4: 256,000+

为什么词汇表越来越大？

多语言分词效率对比：

关键洞察：使用大词汇表的模型（如Qwen 2）在处理中文时更高效。

1.1.3 Embedding：Token如何变成向量

Embedding（向量化）是将Token ID转换为高维向量的过程。

什么是向量？

向量就是一组数字，比如：

[0.1, -0.2, 0.5, 0.8, -0.3, 0.4, ...]  # 4096个数字

为什么需要向量？

Embedding是如何工作的？

Embedding本质上是一个查找表。模型内部存储了一个大矩阵：

词汇表大小: 50000个Token
向量维度: 4096

Embedding矩阵: [50000 × 4096]
    - 第0行: Token ID 0 的向量
    - 第1行: Token ID 1 的向量
    - ...
    - 第49999行: Token ID 49999 的向量

当输入 Token ID = 1234 时：
    → 查找第1234行
    → 返回该行的4096个数字

代码示例：

import torch
import torch.nn as nn

# 参数说明：
# vocab_size: 词汇表大小，即模型能识别的不同Token的数量
# d_model: 模型维度，即每个Token用多少个数字表示
vocab_size = 50000  # 词汇表大小（如GPT-2使用50,257）
d_model = 4096      # 模型维度（如Llama-2-7B使用4,096）

# 创建Embedding层（这就是大模型中的Embedding层）
embedding_layer = nn.Embedding(vocab_size, d_model)

# Token ID → 向量
token_ids = torch.tensor([1234, 5678, 45, 32, 38, 3456])
vectors = embedding_layer(token_ids)

print(vectors.shape)  # torch.Size([6, 4096])
# 6个Token ID，每个变成了4096维的向量

不同模型的Embedding层参数

每个模型都有自己独立的Embedding查找表，参数各不相同：

关键发现：

1. 每个模型都有独立的Embedding查找表：GPT、Llama、Qwen、DeepSeek的Embedding矩阵完全不同
2. 维度选择影响表达能力：维度越高，能表示的语义越丰富，但计算成本也越高
3. 词汇表大小影响分词效率：词汇表越大，分词越细，但Embedding参数也越多

Embedding参数量对比：

洞察：词汇表越大，Embedding参数占比越高。对于小模型（如BERT），Embedding占比可达21%；对于超大模型（如DeepSeek-V3），占比仅0.11%。

位置编码方式对比：

代码示例：不同模型的Embedding层实现：

import torch
import torch.nn as nn

# ===== GPT-2风格：可学习绝对位置编码 =====
class GPT2Embedding(nn.Module):
    def __init__(self, vocab_size=50257, d_model=768, max_seq_len=1024):
        super().__init__()
        # Token Embedding: 将Token ID转换为向量
        self.token_embedding = nn.Embedding(vocab_size, d_model)
        # Position Embedding: 为每个位置学习一个向量
        self.position_embedding = nn.Embedding(max_seq_len, d_model)
    
    def forward(self, input_ids):
        # input_ids: [batch_size, seq_len] - Token ID序列
        seq_len = input_ids.size(1)
        positions = torch.arange(seq_len, device=input_ids.device)
        
        # Token Embedding + Position Embedding
        token_embeds = self.token_embedding(input_ids)      # [batch, seq_len, d_model]
        position_embeds = self.position_embedding(positions) # [seq_len, d_model]
        
        return token_embeds + position_embeds

# ===== BERT风格：Word + Position + Token Type =====
class BertEmbedding(nn.Module):
    def __init__(self, vocab_size=30522, d_model=768, max_seq_len=512, type_vocab_size=2):
        super().__init__()
        # Word Embedding: 将Token ID转换为向量
        self.word_embedding = nn.Embedding(vocab_size, d_model)
        # Position Embedding: 为每个位置学习一个向量
        self.position_embedding = nn.Embedding(max_seq_len, d_model)
        # Token Type Embedding: 区分句子对（如问答任务）
        self.token_type_embedding = nn.Embedding(type_vocab_size, d_model)
    
    def forward(self, input_ids, token_type_ids=None):
        seq_len = input_ids.size(1)
        positions = torch.arange(seq_len, device=input_ids.device)
        
        if token_type_ids is None:
            token_type_ids = torch.zeros_like(input_ids)
        
        # 三种Embedding相加
        word_embeds = self.word_embedding(input_ids)
        position_embeds = self.position_embedding(positions)
        type_embeds = self.token_type_embedding(token_type_ids)
        
        return word_embeds + position_embeds + type_embeds

# ===== Llama/Qwen风格：RoPE位置编码 =====
class LlamaEmbedding(nn.Module):
    def __init__(self, vocab_size=32000, d_model=4096):
        super().__init__()
        # 只有Token Embedding，没有Position Embedding
        # 位置信息通过RoPE在注意力层中编码
        self.token_embedding = nn.Embedding(vocab_size, d_model)
    
    def forward(self, input_ids):
        # 只返回Token Embedding，位置信息在RoPE中处理
        return self.token_embedding(input_ids)

向量的语义特性：

相似含义的Token，向量也相似：

import numpy as np

def cosine_similarity(a, b):
    """计算两个向量的相似度（0~1，越接近1越相似）"""
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# 假设我们有以下向量（简化为3维便于理解）
cat_vector = np.array([0.2, -0.1, 0.5])    # "猫"
kitten_vector = np.array([0.3, -0.2, 0.4]) # "小猫"
car_vector = np.array([-0.5, 0.8, -0.1])   # "汽车"

# 计算相似度
print(cosine_similarity(cat_vector, kitten_vector))  # 0.95 (高，因为都是猫)
print(cosine_similarity(cat_vector, car_vector))      # 0.12 (低，因为语义不同)

总结：

1.1.4 Transformer推理：向量如何被处理

Transformer是大模型的核心架构。它通过自注意力机制处理向量序列。

自注意力的直觉理解：

句子: "小明喜欢小红，因为她很可爱"

当处理"她"时，模型会"关注"句子中的其他词：
- "小红"：高权重（因为"她"指的是"小红"）
- "小明"：低权重
- "喜欢"：低权重
- "可爱"：中等权重

这样模型就能理解"她"指的是谁

Transformer的处理过程：

# 简化的Transformer处理
def transformer_forward(input_vectors):
    """
    input_vectors: [seq_len, d_model]  # 输入向量序列
    output_vectors: [seq_len, d_model]  # 输出向量序列
    """
    # 1. 自注意力：每个向量关注所有其他向量
    attention_output = self_attention(input_vectors)
    
    # 2. 前馈网络：非线性变换
    ff_output = feed_forward(attention_output)
    
    # 3. 残差连接 + 层归一化
    output = layer_norm(input_vectors + ff_output)
    
    return output

# 大模型会重复这个过程32次、64次甚至更多
for i in range(num_layers):
    output_vectors = transformer_forward(output_vectors)

1.1.5 输出层：向量如何变成文字

输出层将Transformer的输出向量转换为每个Token的分数（Logits）。

class OutputLayer(nn.Module):
    def __init__(self, d_model, vocab_size):
        super().__init__()
        self.linear = nn.Linear(d_model, vocab_size)
    
    def forward(self, x):
        # x: [seq_len, d_model]
        # logits: [seq_len, vocab_size]
        logits = self.linear(x)
        return logits

# 假设词汇表大小为50000
output_layer = OutputLayer(d_model=4096, vocab_size=50000)

# Transformer输出
transformer_output = torch.randn(10, 4096)  # 10个Token，每个4096维

# 转换为logits
logits = output_layer(transformer_output)
print(logits.shape)  # torch.Size([10, 50000])
# 每个Token都有50000个分数，对应词汇表中的每个Token

Logits是什么？

Logits是每个Token的"原始分数"，需要经过Softmax转换为概率：

# Logits → 概率
logits = torch.tensor([2.0, 1.0, 0.5, -1.0])
probs = torch.softmax(logits, dim=0)
print(probs)  # tensor([0.57, 0.21, 0.12, 0.10])
# Token 0 的概率是57%，Token 1 是21%，...

1.1.6 解码策略：如何选择下一个Token

从概率分布中选择Token有多种策略：

def greedy_decoding(probs):
    """贪心解码：总是选择概率最高的"""
    return torch.argmax(probs)

def temperature_sampling(probs, temperature=1.0):
    """温度采样：控制随机性"""
    # temperature < 1: 更确定（选择高概率的）
    # temperature > 1: 更随机（给低概率的更多机会）
    adjusted_probs = torch.softmax(torch.log(probs) / temperature, dim=0)
    return torch.multinomial(adjusted_probs, num_samples=1)

def top_k_sampling(probs, k=50):
    """Top-K采样：只在概率最高的K个中选择"""
    top_k_probs, top_k_indices = torch.topk(probs, k)
    # 重新归一化
    top_k_probs = top_k_probs / top_k_probs.sum()
    selected = torch.multinomial(top_k_probs, num_samples=1)
    return top_k_indices[selected]

def top_p_sampling(probs, p=0.9):
    """Top-P采样（核采样）：累积概率达到P时停止"""
    sorted_probs, sorted_indices = torch.sort(probs, descending=True)
    cumulative_probs = torch.cumsum(sorted_probs, dim=0)
    
    # 找到累积概率超过P的位置
    mask = cumulative_probs - sorted_probs >= p
    sorted_probs[mask] = 0
    
    # 重新归一化
    sorted_probs = sorted_probs / sorted_probs.sum()
    selected = torch.multinomial(sorted_probs, num_samples=1)
    return sorted_indices[selected]

解码策略对比：

---

1.2 Embedding到底是什么？

在上一节中，我们多次提到了"Embedding"这个词。但Embedding这个词有两层含义，很多人会混淆。本节将彻底讲清楚。

1.2.1 Embedding的两层含义

Embedding这个词在不同语境下有不同的含义：

关键区别：

Embedding（过程）
    ↓
┌─────────────────────────────────────────────────────────────┐
│ 两种实现方式：                                               │
├─────────────────────────────────────────────────────────────┤
│ 1. 大模型中的Embedding层（模型内部组件）                      │
│    - 是大模型的一部分                                        │
│    - 只能用于Token → 向量                                    │
│    - 向量维度固定（如4096）                                  │
│    - 不能单独使用                                            │
├─────────────────────────────────────────────────────────────┤
│ 2. 独立的Embedding模型（专门的模型）                         │
│    - 是独立的模型                                            │
│    - 可以用于文本 → 向量                                     │
│    - 向量维度可选（如256、1024、3072）                       │
│    - 可以单独使用                                            │
└─────────────────────────────────────────────────────────────┘

1.2.2 大模型中的Embedding层

Embedding层是大模型内部的一个组件，负责将Token ID转换为向量。

class LargeLanguageModel(nn.Module):
    def __init__(self, vocab_size, d_model, num_layers):
        super().__init__()
        
        # Embedding层：Token ID → 向量
        self.token_embedding = nn.Embedding(vocab_size, d_model)
        
        # 位置编码
        self.position_embedding = nn.Embedding(max_seq_len, d_model)
        
        # Transformer层
        self.layers = nn.ModuleList([
            TransformerLayer(d_model) for _ in range(num_layers)
        ])
        
        # 输出层：向量 → Logits
        self.output_layer = nn.Linear(d_model, vocab_size)
    
    def forward(self, input_ids):
        # 1. Embedding：Token ID → 向量
        token_vectors = self.token_embedding(input_ids)
        
        # 2. 添加位置信息
        positions = torch.arange(len(input_ids))
        position_vectors = self.position_embedding(positions)
        vectors = token_vectors + position_vectors
        
        # 3. Transformer处理
        for layer in self.layers:
            vectors = layer(vectors)
        
        # 4. 输出层
        logits = self.output_layer(vectors)
        
        return logits

Embedding层的特点：

1.2.3 独立的Embedding模型

Embedding模型是专门用于生成文本向量的独立模型。

from sentence_transformers import SentenceTransformer

# 加载Embedding模型
model = SentenceTransformer('all-MiniLM-L6-v2')

# 文本 → 向量（直接输入文本，不需要分词）
texts = ["什么是RAG？", "RAG是检索增强生成技术"]
vectors = model.encode(texts)

print(vectors.shape)  # (2, 384)
# 每个文本变成了384维的向量

Embedding模型的特点：

1.2.4 它们的联系与区别

核心区别：

它们的联系：

1. 技术原理相同：都是将离散数据转换为向量
2. 都可以表示语义：相似的文本，向量也相似
3. 可以互相转换：理论上可以用大模型的Embedding层作为Embedding模型

为什么不能直接用大模型的Embedding层作为Embedding模型？

但是，可以用大模型的Embedding层吗？

可以！ 这叫做用大模型作为Embedding模型：

from transformers import AutoModel, AutoTokenizer

# 加载大模型（只使用Embedding层）
tokenizer = AutoTokenizer.from_pretrained("meta-llama/Llama-3-8B")
model = AutoModel.from_pretrained("meta-llama/Llama-3-8B")

# 文本 → Token → Embedding
text = "什么是RAG？"
inputs = tokenizer(text, return_tensors="pt")

# 获取Embedding（不经过Transformer层）
with torch.no_grad():
    # 获取第一层的输出（Embedding层的输出）
    outputs = model(inputs.input_ids, output_hidden_states=True)
    embedding = outputs.hidden_states[0]  # 第0层就是Embedding层

print(embedding.shape)  # torch.Size([1, 6, 4096])

为什么不推荐这样做？

1.2.5 为什么需要独立的Embedding模型？

独立的Embedding模型是专门为"语义表示"设计的，有更好的效果：

主流的Embedding模型：

---

1.3 不同大模型的Embedding层

不同的大模型使用不同的Embedding层设计。了解这些差异有助于理解模型的能力和限制。

1.3.1 主流大模型的Embedding层对比

1.3.2 维度选择的权衡

1.3.3 词汇表大小的影响

---

1.4 Embedding在应用中的作用

Embedding在大模型应用中扮演着关键角色。本节介绍Embedding的主要用途。

1.4.1 语义搜索

语义搜索是Embedding最常用的应用。它通过计算向量相似度来找到语义相关的内容。

from openai import OpenAI
import numpy as np

client = OpenAI()

def get_embedding(text):
    """获取文本的Embedding"""
    response = client.embeddings.create(
        model="text-embedding-3-large",
        input=text
    )
    return response.data[0].embedding

def cosine_similarity(a, b):
    """计算余弦相似度"""
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(a))

# 文档库
documents = [
    "RAG是检索增强生成技术",
    "大语言模型是AI的一种",
    "Python是一种编程语言",
    "机器学习是AI的子领域"
]

# 查询
query = "什么是RAG？"

# 获取Embedding
query_embedding = get_embedding(query)
doc_embeddings = [get_embedding(doc) for doc in documents]

# 计算相似度
similarities = [cosine_similarity(query_embedding, doc_emb) for doc_emb in doc_embeddings]

# 找到最相似的文档
most_similar_idx = np.argmax(similarities)
print(f"最相关的文档: {documents[most_similar_idx]}")
print(f"相似度: {similarities[most_similar_idx]:.4f}")

1.4.2 RAG（检索增强生成）

RAG是Embedding和大模型协作的典型应用：

用户问题: "什么是RAG？"
        ↓
┌─────────────────────────────────────────────────────────────┐
│ 第1步：Embedding模型                                        │
│ 将问题转为向量: [0.1, 0.2, 0.3, ..., 0.4]                  │
└─────────────────────────────────────────────────────────────┘
        ↓
┌─────────────────────────────────────────────────────────────┐
│ 第2步：向量检索                                             │
│ 在知识库中找到最相似的文档                                   │
└─────────────────────────────────────────────────────────────┘
        ↓
┌─────────────────────────────────────────────────────────────┐
│ 第3步：大模型生成                                           │
│ 基于检索到的文档，生成回答                                   │
└─────────────────────────────────────────────────────────────┘
        ↓
模型输出: "RAG是检索增强生成技术..."

代码示例：

from openai import OpenAI
import numpy as np

client = OpenAI()

def rag_answer(question, documents):
    """RAG问答"""
    
    # 1. 获取问题的Embedding
    question_embedding = get_embedding(question)
    
    # 2. 获取文档的Embedding
    doc_embeddings = [get_embedding(doc) for doc in documents]
    
    # 3. 找到最相关的文档
    similarities = [cosine_similarity(question_embedding, doc_emb) for doc_emb in doc_embeddings]
    most_similar_idx = np.argmax(similarities)
    relevant_doc = documents[most_similar_idx]
    
    # 4. 使用大模型生成回答
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "基于以下上下文回答问题。"},
            {"role": "user", "content": f"上下文：{relevant_doc}\n\n问题：{question}"}
        ]
    )
    
    return response.choices[0].message.content

# 使用
documents = [
    "RAG是检索增强生成技术，它结合了检索和生成...",
    "大语言模型是AI的一种，它通过预训练学习语言...",
    "Python是一种编程语言，广泛用于AI开发..."
]

answer = rag_answer("什么是RAG？", documents)
print(answer)

1.4.3 文本分类

文本分类是将文本分为不同类别的任务。Embedding可以将文本转换为向量，然后用分类器进行分类。

from sklearn.linear_model import LogisticRegression
from openai import OpenAI
import numpy as np

client = OpenAI()

# 训练数据
train_texts = [
    "这部电影太好看了",
    "剧情很无聊",
    "演员演技很棒",
    "浪费时间",
    "推荐大家去看",
    "不值得票价"
]
train_labels = [1, 0, 1, 0, 1, 0]  # 1=正面, 0=负面

# 获取Embedding
train_embeddings = [get_embedding(text) for text in train_texts]

# 训练分类器
classifier = LogisticRegression()
classifier.fit(train_embeddings, train_labels)

# 预测新文本
new_text = "这个电影还不错"
new_embedding = get_embedding(new_text)
prediction = classifier.predict([new_embedding])
print(f"预测结果: {'正面' if prediction[0] == 1 else '负面'}")

1.4.4 文本聚类

文本聚类是将相似文本分组的任务。Embedding可以将文本转换为向量，然后用聚类算法进行分组。

from sklearn.cluster import KMeans
from openai import OpenAI
import numpy as np

client = OpenAI()

# 文本列表
texts = [
    "苹果发布了新iPhone",
    "华为发布了新手机",
    "今天天气很好",
    "明天会下雨",
    "iPhone销量创新高",
    "气温将达到30度"
]

# 获取Embedding
embeddings = [get_embedding(text) for text in texts]

# 聚类
kmeans = KMeans(n_clusters=2, random_state=0)
clusters = kmeans.fit_predict(embeddings)

# 输出聚类结果
for i, (text, cluster) in enumerate(zip(texts, clusters)):
    print(f"文本: {text} → 类别: {cluster}")
# 文本: 苹果发布了新iPhone → 类别: 0
# 文本: 华为发布了新手机 → 类别: 0
# 文本: 今天天气很好 → 类别: 1
# 文本: 明天会下雨 → 类别: 1
# 文本: iPhone销量创新高 → 类别: 0
# 文本: 气温将达到30度 → 类别: 1

---

1.5 总结：大模型与Embedding的关系

1.5.1 关键概念回顾

1.5.2 它们如何协作

┌─────────────────────────────────────────────────────────────┐
│ 应用场景：RAG                                                │
├─────────────────────────────────────────────────────────────┤
│ 1. Embedding模型：将文档和问题转为向量                       │
│ 2. 向量数据库：存储和检索向量                                │
│ 3. 大模型：基于检索结果生成回答                              │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│ 应用场景：对话                                               │
├─────────────────────────────────────────────────────────────┤
│ 1. 大模型内部的Embedding层：将输入Token转为向量              │
│ 2. Transformer：处理向量，理解语义                           │
│ 3. 输出层 + Decoding：生成回答                               │
└─────────────────────────────────────────────────────────────┘

1.5.3 给开发者的建议

---

二、大模型应用概述

传统AI应用开发遵循"任务特定"范式：为每个任务训练专门的模型，需要大量标注数据和特征工程。大语言模型的出现带来了根本性转变：

2.1 LLM应用的核心特征

LLM应用与传统软件应用有显著差异，主要体现在以下四个维度：

2.2 2026年应用生态全景图

经过两年多的发展，LLM应用生态已经形成清晰的分层架构：

┌─────────────────────────────────────────────────────────────┐
│                     应用层 (Applications)                    │
│   Chatbot · 代码助手 · 内容创作 · 数据分析 · 游戏NPC ...    │
├─────────────────────────────────────────────────────────────┤
│                   编排层 (Orchestration)                      │
│        LangChain · LangGraph · Dify · Flowise · n8n         │
├─────────────────────────────────────────────────────────────┤
│                    协议层 (Protocol)                          │
│              MCP (Model Context Protocol)                    │
├─────────────────────────────────────────────────────────────┤
│                 能力扩展层 (Capabilities)                     │
│           RAG · Agent · Skill · Memory · Tools              │
├─────────────────────────────────────────────────────────────┤
│                基础设施层 (Infrastructure)                    │
│    向量数据库 · 可观测性 · 模型路由 · 安全防护 · 成本控制    │
├─────────────────────────────────────────────────────────────┤
│                    模型层 (Models)                            │
│     OpenAI · Claude · DeepSeek · Qwen · Llama · 本地模型     │
└─────────────────────────────────────────────────────────────┘

按成熟度划分，当前LLM应用可分为三个梯队：

---

三、RAG：检索增强生成

3.1 RAG的核心原理

RAG（Retrieval-Augmented Generation） 是将LLM与外部知识库结合的技术。其核心思想是：在LLM生成回答之前，先从知识库中检索相关文档，将其作为上下文注入提示词。

为什么需要RAG？

RAG的工作流程：

用户问题
    ↓
┌─────────────┐
│  查询理解   │ → 问题改写、查询扩展
└─────────────┘
    ↓
┌─────────────┐
│   检索器    │ → 向量检索 + 关键词检索
└─────────────┘
    ↓
┌─────────────┐
│   重排序器  │ → Cross-Encoder精排
└─────────────┘
    ↓
┌─────────────┐
│  上下文构建  │ → 组装提示词
└─────────────┘
    ↓
┌─────────────┐
│   LLM生成   │ → 基于上下文生成回答
└─────────────┘
    ↓
  最终回答

3.2 混合检索与重排序

混合检索结合了向量检索（语义匹配）和关键词检索（精确匹配），能显著提升检索质量：

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain.vectorstores import Chroma

# BM25检索器（关键词匹配）
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 50

# 向量检索器（语义匹配）
vectorstore = Chroma.from_documents(documents, embedding_model)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 50})

# 混合检索（RRF融合）
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]  # BM25权重0.4，向量权重0.6
)

重排序是检索后的精排步骤，使用Cross-Encoder模型对检索结果进行重新排序：

from langchain.retrievers import ContextualCompressionRetriever
from langchain_cohere import CohereRerank

# Cohere Rerank 3.5
compressor = CohereRerank(model="rerank-v3.5", top_n=5)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=ensemble_retriever
)

3.3 如何自己开发一个RAG系统

步骤1：文档加载与分块

from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 加载文档
loader = DirectoryLoader("./docs", glob="**/*.md", loader_cls=TextLoader)
documents = loader.load()

# 分块
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", " ", ""]
)
chunks = text_splitter.split_documents(documents)

步骤2：向量化与存储

from langchain_openai import OpenAIEmbeddings
from langchain.vectorstores import Chroma

# 初始化Embedding模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")

# 创建向量数据库
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 创建检索器
retriever = vectorstore.as_retriever(
    search_type="mmr",  # 最大边际相关性
    search_kwargs={"k": 5}
)

步骤3：构建RAG链

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# 提示词模板
template = """
基于以下上下文回答问题。如果上下文中没有相关信息，请说"我不知道"。

上下文：
{context}

问题：
{question}

回答：
"""

prompt = ChatPromptTemplate.from_template(template)

# LLM
llm = ChatOpenAI(model="gpt-4o")

# RAG链
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# 使用
answer = rag_chain.invoke("什么是RAG？")

---

四、Agent：智能体系统

4.1 Agent的核心原理

Agent是能自主感知环境、做出决策、执行行动的AI系统。与传统的LLM调用不同，Agent具有自主性和目标导向性。

Agent的核心组件：

ReAct架构是最经典的Agent架构，通过"思考-行动-观察"循环解决问题：

Thought: 我需要搜索今天的天气
Action: search_weather("北京")
Observation: 北京今天晴，25°C
Thought: 我已经知道天气了，可以回答用户
Action: respond("北京今天晴，25°C")

4.2 多智能体协作

多智能体系统让多个Agent协作完成复杂任务：

4.3 如何自己开发一个Agent

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate

# 定义工具
@tool
def search_database(query: str) -> str:
    """搜索数据库中的信息"""
    return f"搜索结果：{query}"

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件"""
    return f"邮件已发送到 {to}"

# LLM
llm = ChatOpenAI(model="gpt-4o")

# 提示词
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有用的助手，可以使用工具来完成任务。"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

# 创建Agent
agent = create_react_agent(llm, [search_database, send_email], prompt)

# 创建执行器
executor = AgentExecutor(
    agent=agent,
    tools=[search_database, send_email],
    verbose=True,
    max_iterations=10
)

# 运行
result = executor.invoke({"input": "帮我搜索数据库中的用户信息"})

---

五、MCP：模型上下文协议

5.1 MCP的核心原理

MCP（Model Context Protocol） 是由Anthropic于2024年11月发布的开放协议，定义了AI应用如何连接外部工具和数据源。

MCP架构：

┌─────────────────┐      ┌─────────────────┐
│   MCP Client    │ ←──→ │   MCP Server    │
│  (AI应用)       │      │  (工具/数据源)  │
└─────────────────┘      └─────────────────┘

三种能力类型：

5.2 如何自己开发一个MCP服务器

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";

const server = new McpServer({
  name: "my-mcp-server",
  version: "1.0.0"
});

// 定义工具
server.tool(
  "calculate",
  { expression: z.string() },
  async ({ expression }) => {
    try {
      const result = eval(expression);
      return {
        content: [{ type: "text", text: String(result) }]
      };
    } catch (error) {
      return {
        content: [{ type: "text", text: `错误: ${error.message}` }],
        isError: true
      };
    }
  }
);

// 启动服务器
const transport = new StdioServerTransport();
await server.connect(transport);

---

六、Harness工程与性能优化

6.1 Harness工程概述

Harness工程是围绕LLM调用构建的外部编排层，包括系统提示、工具选择、执行流、记忆系统和中间件钩子。

Harness的三个核心设计维度：

6.2 Token压缩：Headroom

Headroom是一个上下文优化层，在LLM调用前压缩所有输入内容。

三阶段压缩流水线：

Stage 1: CacheAligner → 稳定消息前缀，使KV Cache命中
Stage 2: ContentRouter → 自动检测内容类型，路由到最优压缩器
Stage 3: IntelligentContext → 按重要性评分，删除低价值消息

6种压缩算法：

6.3 模型路由

模型路由根据请求特征自动选择最合适的模型：

class ModelRouter:
    """基于规则的模型路由"""
    
    def route(self, query: str) -> str:
        complexity = self.classify_complexity(query)
        
        if complexity == "simple":
            return "gpt-4o-mini"  # 便宜
        elif complexity == "medium":
            return "gpt-4o"  # 平衡
        else:
            return "claude-opus-4"  # 强大

6.4 缓存策略

6.5 流式输出

流式输出基于**Server-Sent Events（SSE）**协议：

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json

app = FastAPI()

async def generate_tokens(prompt: str):
    response = await openai_client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        stream=True
    )
    
    async for chunk in response:
        if chunk.choices[0].delta.content:
            yield f"data: {json.dumps({'content': chunk.choices[0].delta.content})}\n\n"
    
    yield "data: [DONE]\n\n"

@app.post("/chat/stream")
async def chat_stream(prompt: str):
    return StreamingResponse(
        generate_tokens(prompt),
        media_type="text/event-stream"
    )

6.6 错误处理与重试

错误分类：

---

七、Agent Skills系统

7.1 Skills的核心原理

Skills是一种模块化、文件系统抽象，让Agent按需加载领域专业知识。

SKILL.md文件格式：

---
name: code-review
description: |
  代码审查技能，用于审查代码质量、安全性和最佳实践。
  
  触发条件：
  - 当用户请求审查代码时
  - 当PR需要审查时
  
allowed-tools:
  - read
  - lsp_diagnostics
  - grep
---

# 代码审查技能

## 审查维度

| 维度 | 检查项 |
|------|--------|
| **代码质量** | 命名规范、代码结构、注释质量 |
| **安全性** | 输入验证、SQL注入防护、XSS防护 |
| **性能** | 算法复杂度、资源使用、缓存策略 |

## 审查流程

1. 使用`read`工具读取代码文件
2. 使用`lsp_diagnostics`检查语法错误
3. 使用`grep`搜索潜在问题模式
4. 生成审查报告

---

八、LLM Gateway与可观测性

8.1 LLM Gateway原理

LLM Gateway是位于应用和LLM提供商之间的控制平面：

代表产品对比：

8.2 AI可观测性

AI可观测性通过捕获每条trace、token、成本和评估分数，让团队看到AI做了什么以及为什么。

代表产品对比：

---

九、向量数据库与Embedding

9.1 向量数据库原理

向量数据库是RAG应用的核心基础设施，存储和检索高维向量嵌入。

代表产品对比：

9.2 Embedding模型

Embedding模型将文本/图像转换为高维向量表示。

代表模型对比：

---

十、垂直领域应用

10.1 代码生成与软件开发

10.2 内容创作与媒体

10.3 医疗健康

10.4 金融领域

10.5 教育领域

---

十一、游戏领域的LLM应用

11.1 NPC智能交互

NVIDIA ACE Game Agent SDK：

实际应用案例：

11.2 剧情与内容生成

11.3 案例研究

Maicraft：LLM驱动的Minecraft代理

AgentGal：开放世界多Agent角色扮演

---

十二、未来趋势与挑战

12.1 技术瓶颈

12.2 伦理与安全

12.3 Agentic AI的未来

根据Forrester 2026年报告：

未来趋势：

---

总结

2026年，大模型应用开发已经从"能用"进入"好用"的工程化阶段。RAG、Agent、MCP、Skill等技术栈日趋成熟，游戏、医疗、金融、教育等垂直领域应用蓬勃发展。

给开发者的建议：

---

参考资源

官方文档

• LangChain文档
• LangGraph文档
• MCP规范
• Agent Skills规范

开源项目

• LangChain
• LangGraph
• Dify
• RAGFlow
• ECC

行业报告

• Forrester: The State of Agentic AI 2026
• Gartner: AI Agent预测
• Deloitte: Agentic AI Strategy