构建AI智能体：Encoder-only与Decoder-only模型架构：基于本地小模型的实践解析

from transformers import BertTokenizer, BertForSequenceClassification
from modelscope import snapshot_download
import torch
from torch.nn.functional import cosine_similarity

# 1. 加载模型和分词器
cache_dir = "D:\\modelscope\\hub"
model_name = "google-bert/bert-base-chinese"
local_model_path = snapshot_download(model_name, cache_dir=cache_dir)
tokenizer = BertTokenizer.from_pretrained(local_model_path)
# 对于相似度任务，我们也可以使用专门为句向量优化的模型，如 sentence-transformers
# 这里我们用基础BERT演示获取句向量的方法
model = BertForSequenceClassification.from_pretrained(local_model_path)
model.eval()

# 2. 准备句子对
sentence_pairs = [
    ("今天天气怎么样？", "我想知道今天的天气状况。"), # 语义相似
    ("苹果是一种水果。", "特斯拉是电动汽车公司。"), # 语义不相似
    ("深度学习需要大量数据。", "机器学习对数据量有要求。") # 语义相似
]

def get_sentence_embedding(text, model, tokenizer):
    """获取句子的向量表示（使用 [CLS] 标记的输出）"""
    inputs = tokenizer(text, return_tensors="pt", padding=True, truncation=True, max_length=128)
    with torch.no_grad():
        outputs = model(**inputs, output_hidden_states=True)
    # 取最后一层隐藏状态中 [CLS] 标记对应的向量作为句向量
    last_hidden_states = outputs.hidden_states[-1]
    cls_embedding = last_hidden_states[:, 0, :] # [batch_size, hidden_size]
    return cls_embedding

# 3. 计算并比较相似度
print("句子对相似度比较:")
print("=" * 60)

for sent1, sent2 in sentence_pairs:
    emb1 = get_sentence_embedding(sent1, model, tokenizer)
    emb2 = get_sentence_embedding(sent2, model, tokenizer)
    
    # 计算余弦相似度
    sim = cosine_similarity(emb1, emb2).item()
    
    print(f"句子 A: 「{sent1}」")
    print(f"句子 B: 「{sent2}」")
    print(f"余弦相似度: {sim:.4f}")
    if sim > 0.7:
        print("结论: 语义相似")
    else:
        print("结论: 语义不相似")
    print("-" * 50)

from transformers import BertTokenizer, BertForTokenClassification
from modelscope import snapshot_download
import torch
from torch.nn.functional import cosine_similarity

# 1. 加载模型和分词器
cache_dir = "D:\\modelscope\\hub"
model_name = "google-bert/bert-base-chinese"
local_model_path = snapshot_download(model_name, cache_dir=cache_dir)
tokenizer = BertTokenizer.from_pretrained(local_model_path)
# 对于相似度任务，我们也可以使用专门为句向量优化的模型，如 sentence-transformers

# 假设我们定义了一个标签集：O（非实体），B-PER（人名开始），I-PER（人名内部），B-ORG（组织开始），I-ORG（组织内部），B-LOC（地点开始），I-LOC（地点内部）
label_list = ['O', 'B-PER', 'I-PER', 'B-ORG', 'I-ORG', 'B-LOC', 'I-LOC']
model = BertForTokenClassification.from_pretrained(local_model_path, num_labels=len(label_list))
model.eval()

# 2. 待分析的文本
text = "阿里巴巴的总部位于中国杭州市，马云是它的创始人之一。"

# 3. 预处理
inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=128)
tokens = tokenizer.convert_ids_to_tokens(inputs['input_ids'][0]) # 将ID转换回token，便于查看

# 4. 模型预测
with torch.no_grad():
    outputs = model(**inputs)

logits = outputs.logits
predictions = torch.argmax(logits, dim=-1)[0].tolist() # 获取每个token的预测标签ID

# 5. 对齐和输出结果
print(f"文本: 「{text}」")
print("\n实体识别结果:")
print("Token\t\t\t预测标签")
print("-" * 40)

for token, prediction_id in zip(tokens, predictions):
    # 跳过特殊的 [CLS] 和 [SEP] 等token
    if token in ['[CLS]', '[SEP]', '[PAD]']:
        continue
    label = label_list[prediction_id]
    # 处理中文子词（以 ## 开头）
    if token.startswith('##'):
        print(f"{token[2:]}\t\t\t{label}") # 去掉 ##，与上一个token连接
    else:
        print(f"{token}\t\t\t{label}")  

from transformers import BertTokenizer, BertForSequenceClassification
from modelscope import snapshot_download
import torch
from torch.nn.functional import cosine_similarity

# 1. 加载模型和分词器
cache_dir = "D:\\modelscope\\hub"
model_name = "google-bert/bert-base-chinese"
local_model_path = snapshot_download(model_name, cache_dir=cache_dir)
tokenizer = BertTokenizer.from_pretrained(local_model_path)
# 对于相似度任务，我们也可以使用专门为句向量优化的模型，如 sentence-transformers
# 这里我们用基础BERT演示获取句向量的方法
model = BertForSequenceClassification.from_pretrained(local_model_path)
model.eval()

# 注意：对于具体任务，我们通常会加载一个在特定任务上微调过的模型。
# 这里为了演示，我们使用基础模型，但正常情况下你应该使用任务特定的模型。
model = BertForSequenceClassification.from_pretrained(local_model_path, num_labels=2) # 假设2分类：正面/负面
model.eval() # 设置为评估模式

# 2. 待分类的文本
texts = [
    "这部电影真是太精彩了，演员演技炸裂！",
    "剧情拖沓，逻辑混乱，浪费了我两个小时。",
    "中规中矩，没什么特别的亮点。"
]

# 3. 预处理和模型预测
for text in texts:
    # 编码文本，返回PyTorch张量
    inputs = tokenizer(text, return_tensors="pt", padding=True, truncation=True, max_length=128)
    
    # 不计算梯度，加快推理速度
    with torch.no_grad():
        outputs = model(**inputs)
    
    # 获取预测结果
    # logits是模型最后的原始输出
    logits = outputs.logits
    # 使用softmax将logits转换为概率
    probabilities = torch.nn.functional.softmax(logits, dim=-1)
    # 获取预测的类别（概率最大的那个）
    predicted_class_id = torch.argmax(probabilities, dim=-1).item()
    
    # 打印结果
    sentiment = "正面" if predicted_class_id == 1 else "负面"
    confidence = probabilities[0][predicted_class_id].item()
    
    print(f"文本: 「{text}」")
    print(f"  情感: {sentiment} (置信度: {confidence:.4f})")
    print("-" * 50)        

import torch
from transformers import AutoModel, AutoTokenizer
import matplotlib.pyplot as plt
import seaborn as sns
import numpy as np
from modelscope import snapshot_download

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

class AttentionVisualizer:
    cache_dir = "D:\\modelscope\\hub"
    model_name = "google-bert/bert-base-chinese"
    local_model_path = snapshot_download(model_name, cache_dir=cache_dir)
    def __init__(self, model_name="bert-base-chinese"):
        self.tokenizer = AutoTokenizer.from_pretrained(self.local_model_path)
        self.model = AutoModel.from_pretrained(self.local_model_path, output_attentions=True)
        self.model.eval()
    
    def visualize_attention(self, text, layer=0, head=0):
        """可视化指定层和头的注意力模式"""
        # 编码文本
        inputs = self.tokenizer(text, return_tensors='pt', padding=True, truncation=True)
        tokens = self.tokenizer.convert_ids_to_tokens(inputs['input_ids'][0])
        
        # 前向传播获取注意力
        with torch.no_grad():
            outputs = self.model(**inputs)
            attentions = outputs.attentions  # 所有层的注意力权重
        
        # 获取指定层和头的注意力权重
        attention_weights = attentions[layer][0, head].numpy()
        
        # 创建热力图
        plt.figure(figsize=(10, 8))
        sns.heatmap(attention_weights, 
                   xticklabels=tokens,
                   yticklabels=tokens,
                   cmap='Blues',
                   annot=False,
                   cbar=True)
        plt.title(f'BERT 注意力机制 - 第{layer}层第{head}头\n文本: "{text}"')
        plt.xlabel('Key Tokens')
        plt.ylabel('Query Tokens')
        plt.xticks(rotation=45)
        plt.yticks(rotation=0)
        plt.tight_layout()
        plt.show()
        
        return attention_weights, tokens

# 使用示例
visualizer = AttentionVisualizer()
text = "今天天气很好"
attention_weights, tokens = visualizer.visualize_attention(text)

print("输入文本:", text)
print("分词结果:", tokens)
print("注意力权重形状:", attention_weights.shape)
print("\n注意力模式解释:")
for i, token in enumerate(tokens):
    attention_scores = attention_weights[i]
    top_indices = np.argsort(attention_scores)[-3:][::-1]  # 取前3个最关注的token
    top_tokens = [tokens[idx] for idx in top_indices]
    top_scores = [attention_scores[idx] for idx in top_indices]
    print(f"{token} → {list(zip(top_tokens, top_scores))}")

def analyze_attention_patterns():
    """分析不同层的注意力模式"""
    visualizer = AttentionVisualizer()
    text = "今天天气很好适合出门"
    
    inputs = visualizer.tokenizer(text, return_tensors='pt')
    tokens = visualizer.tokenizer.convert_ids_to_tokens(inputs['input_ids'][0])
    
    with torch.no_grad():
        outputs = visualizer.model(**inputs)
        attentions = outputs.attentions
    
    print("BERT模型注意力模式分析")
    print("=" * 50)
    print(f"输入文本: {text}")
    print(f"分词: {tokens}")
    print(f"总层数: {len(attentions)}")
    print(f"每层头数: {attentions[0].shape[1]}")
    print(f"序列长度: {attentions[0].shape[2]}")
    print()
    
    # 分析第一层第一个头的注意力模式
    first_layer_attention = attentions[0][0, 0]  # [batch, head, query, key]
    
    print("第一层第一个头的注意力模式:")
    print("Query → [最关注的Key Tokens]")
    print("-" * 40)
    
    for i in range(len(tokens)):
        query_token = tokens[i]
        attention_scores = first_layer_attention[i]
        
        # 找出注意力权重最高的几个token
        significant_attention = [(tokens[j], attention_scores[j].item()) 
                               for j in range(len(tokens)) 
                               if attention_scores[j] > 0.1]
        
        # 按注意力权重排序
        significant_attention.sort(key=lambda x: x[1], reverse=True)
        
        if significant_attention:
            targets = [f"{token}({score:.3f})" for token, score in significant_attention]
            print(f"{query_token:8} → {targets}")

# 运行分析
analyze_attention_patterns()

def visualize_multiple_heads(text="今天天气很好", layer=0):
    """可视化同一层的多个注意力头"""
    visualizer = AttentionVisualizer()
    inputs = visualizer.tokenizer(text, return_tensors='pt')
    tokens = visualizer.tokenizer.convert_ids_to_tokens(inputs['input_ids'][0])
    
    with torch.no_grad():
        outputs = visualizer.model(**inputs)
        attentions = outputs.attentions
    
    # 获取指定层的所有头
    layer_attention = attentions[layer][0]  # [num_heads, seq_len, seq_len]
    num_heads = layer_attention.shape[0]
    
    print(f"第{layer}层的{num_heads}个注意力头模式:")
    print("=" * 60)
    
    # 显示前4个头
    fig, axes = plt.subplots(2, 2, figsize=(15, 12))
    axes = axes.flatten()
    
    for head in range(min(4, num_heads)):
        attention_weights = layer_attention[head].numpy()
        
        sns.heatmap(attention_weights,
                   xticklabels=tokens,
                   yticklabels=tokens,
                   cmap='Blues',
                   ax=axes[head],
                   cbar=True)
        axes[head].set_title(f'头 {head} 注意力模式')
        axes[head].set_xlabel('Key Tokens')
        axes[head].set_ylabel('Query Tokens')
    
    plt.tight_layout()
    plt.show()
    
    # 文本输出每个头的模式
    for head in range(min(4, num_heads)):
        print(f"\n头 {head} 的注意力模式:")
        attention_weights = layer_attention[head]
        for i, token in enumerate(tokens):
            scores = attention_weights[i]
            max_idx = torch.argmax(scores).item()
            print(f"  {token} → 最关注: {tokens[max_idx]}({scores[max_idx]:.3f})")

# 可视化多头注意力
visualize_multiple_heads()

from transformers import AutoModelForCausalLM, AutoTokenizer
import torch
from modelscope import snapshot_download

class DecoderOnlyExample:
    def __init__(self):
        # 使用ModelScope中的GPT系列小模型
        cache_dir = "D:\\modelscope\\hub"
        self.model_name = "Fengshenbang/Wenzhong-GPT2-110M-chinese-v2"
        local_model_path = snapshot_download(self.model_name, cache_dir=cache_dir)
        self.tokenizer = AutoTokenizer.from_pretrained(local_model_path)
        self.model = AutoModelForCausalLM.from_pretrained(local_model_path)
        
        # 设置padding token
        if self.tokenizer.pad_token is None:
            self.tokenizer.pad_token = self.tokenizer.eos_token
    
    def generate_text(self, prompt, max_length=100, temperature=0.8):
        """文本生成示例"""
        inputs = self.tokenizer(prompt, return_tensors='pt')
        
        with torch.no_grad():
            outputs = self.model.generate(
                inputs.input_ids,
                max_length=max_length,
                num_return_sequences=1,
                temperature=temperature,
                do_sample=True,
                pad_token_id=self.tokenizer.eos_token_id,
                no_repeat_ngram_size=3
            )
        
        generated_text = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
        return generated_text
    
    def calculate_perplexity(self, text):
        """计算文本困惑度"""
        inputs = self.tokenizer(text, return_tensors='pt')
        with torch.no_grad():
            outputs = self.model(**inputs, labels=inputs.input_ids)
            loss = outputs.loss
            perplexity = torch.exp(loss)
        return perplexity.item()

# 使用示例
decoder_example = DecoderOnlyExample()
prompt = "人工智能的未来发展"
generated_text = decoder_example.generate_text(prompt)
print(f"输入提示: {prompt}")
print(f"生成的文本: {generated_text}")


def simple_causal_attention_demo():
    print("\n" + "="*60)
    """最简单的因果注意力演示"""
    print("因果注意力模式 (Decoder-only)")
    print("=" * 40)
    
    # 输入序列
    sequence = ["<s>", "今天", "天气", "很"]
    print(f"输入序列: {' '.join(sequence)}")
    print()
    
    print("注意力模式:")
    print("-" * 25)
    
    # 为每个位置显示可以关注的位置
    for i in range(len(sequence)):
        current_token = sequence[i]
        # 当前位置可以关注的所有位置（从0到i）
        can_attend = sequence[:i+1]
        
        print(f"{current_token:4} → {', '.join(can_attend)}")

# 运行演示
simple_causal_attention_demo()

def clear_causal_demo():
    """更清晰的因果注意力演示"""
    print("\n" + "="*60)
    print("Decoder-only 因果注意力")
    print("=" * 35)
    
    tokens = ["<s>", "今天", "天气", "很", "好"]
    
    print("规则: 每个token只能关注它自己和前面的所有token")
    print()
    
    for i in range(len(tokens)):
        current = tokens[i]
        previous = tokens[:i+1]  # 从开始到当前的所有token
        
        arrow = "→"
        if i == 0:
            arrow = "→ (只能关注自己)"
        elif i == len(tokens) - 1:
            arrow = "→ (可以关注前面所有token)"
        
        print(f"{current:4} {arrow:20} {', '.join(previous)}")

# 运行清晰版本
clear_causal_demo()

def minimal_demo():
    print("\n" + "="*60)
    """最简化的核心逻辑演示"""
    tokens = ["起始", "今天", "天气", "很", "好"]
    
    print("因果注意力核心规则:")
    print("每个位置只能看到它和它之前的所有位置")
    print()
    
    for i in range(len(tokens)):
        can_see = tokens[:i+1]
        print(f"生成 '{tokens[i]}' 时，模型能看到: {can_see}")

# 运行最简版本
minimal_demo()

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
import numpy as np
from modelscope import snapshot_download

class CausalAttentionVisualizer:
    def __init__(self, model_name="gpt2-chinese-cluecorpussmall"):
        cache_dir = "D:\\modelscope\\hub"
        self.model_name = "Fengshenbang/Wenzhong-GPT2-110M-chinese-v2"
        local_model_path = snapshot_download(self.model_name, cache_dir=cache_dir)
        self.tokenizer = AutoTokenizer.from_pretrained(local_model_path)
        self.model = AutoModelForCausalLM.from_pretrained(local_model_path, output_attentions=True)
        self.model.eval()
        
        if self.tokenizer.pad_token is None:
            self.tokenizer.pad_token = self.tokenizer.eos_token
    
    def get_causal_attention_mask(self, seq_len):
        """生成因果注意力掩码"""
        # 创建下三角矩阵，1表示允许关注，0表示不允许
        mask = torch.tril(torch.ones(seq_len, seq_len))
        return mask
    
    def visualize_causal_pattern(self, text):
        """可视化因果注意力模式"""
        # 编码文本
        inputs = self.tokenizer(text, return_tensors='pt')
        seq_len = inputs['input_ids'].shape[1]
        
        print("因果注意力模式分析")
        print("=" * 40)
        print(f"序列长度: {seq_len}")
        print(f"输入序列: {text}")
        print("\n注意力模式:")
        print("-" * 30)
        
        # 生成因果注意力掩码
        causal_mask = self.get_causal_attention_mask(seq_len)
        
        # 输出注意力模式
        for i in range(seq_len):
            # 获取当前位置可以关注的token索引
            allowed_indices = torch.where(causal_mask[i] == 1)[0].tolist()
            
            if i == 0:
                # 第一个token只能关注自己
                print(f"位置 {i} → [位置 {i}]")
            else:
                # 其他位置可以关注从0到当前位置的所有token
                positions = [f"位置 {j}" for j in allowed_indices]
                print(f"位置 {i} → {positions}")
        
        return causal_mask

# 使用示例
visualizer = CausalAttentionVisualizer()
text = "今天天气很"
causal_mask = visualizer.visualize_causal_pattern(text)

def analyze_causal_attention_detailed():
    """详细分析因果注意力模式"""
    visualizer = CausalAttentionVisualizer()
    text = "今天天气很好适合出门"
    
    inputs = visualizer.tokenizer(text, return_tensors='pt')
    seq_len = inputs['input_ids'].shape[1]
    
    print("Decoder-only 因果注意力机制")
    print("=" * 50)
    print(f"输入序列长度: {seq_len}")
    print("\n完整的注意力模式矩阵:")
    print("-" * 40)
    
    # 生成因果注意力掩码
    causal_mask = visualizer.get_causal_attention_mask(seq_len)
    
    # 输出矩阵形式的注意力模式
    print("   ", end="")
    for j in range(seq_len):
        print(f"{j:2} ", end="")
    print()
    
    for i in range(seq_len):
        print(f"{i:2} ", end="")
        for j in range(seq_len):
            if causal_mask[i, j] == 1:
                print(" ● ", end="")
            else:
                print(" · ", end="")
        print(f"  位置 {i} 可以关注: {[k for k in range(seq_len) if causal_mask[i, k] == 1]}")
    
    print("\n其中: ● = 允许关注, · = 禁止关注")

# 运行详细分析
analyze_causal_attention_detailed()

def verify_model_attention():
    """验证实际模型的因果注意力"""
    visualizer = CausalAttentionVisualizer()
    text = "今天天气很"
    
    inputs = visualizer.tokenizer(text, return_tensors='pt')
    
    with torch.no_grad():
        outputs = visualizer.model(**inputs, output_attentions=True)
    
    attentions = outputs.attentions  # 所有层的注意力
    seq_len = inputs['input_ids'].shape[1]
    
    print("实际模型因果注意力验证")
    print("=" * 40)
    print(f"模型层数: {len(attentions)}")
    print(f"每层头数: {attentions[0].shape[1]}")
    print(f"序列长度: {seq_len}")
    
    # 检查第一层第一个头的注意力
    first_head_attention = attentions[0][0, 0]  # [seq_len, seq_len]
    
    print("\n第一层第一个头注意力模式:")
    print("位置 → [可以关注的位置] (注意力权重>0.01)")
    print("-" * 50)
    
    for i in range(seq_len):
        # 找出有显著注意力权重的位置
        significant_positions = []
        for j in range(seq_len):
            if first_head_attention[i, j] > 0.01:
                significant_positions.append(j)
        
        print(f"位置 {i} → 位置 {significant_positions}")

# 验证模型注意力
verify_model_attention()

import matplotlib.pyplot as plt

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

def plot_causal_attention_matrix(seq_len=6):
    """绘制因果注意力矩阵图"""
    # 创建因果注意力掩码
    causal_mask = torch.tril(torch.ones(seq_len, seq_len))
    
    plt.figure(figsize=(10, 8))
    plt.imshow(causal_mask, cmap='Blues', aspect='auto')
    
    # 添加网格和标签
    for i in range(seq_len):
        for j in range(seq_len):
            color = 'white' if causal_mask[i, j] > 0.5 else 'black'
            plt.text(j, i, '●' if causal_mask[i, j] > 0.5 else '·', 
                    ha='center', va='center', color=color, fontsize=20)
    
    plt.xlabel('Key Position (被关注的位置)')
    plt.ylabel('Query Position (关注的位置)')
    plt.title('Decoder-only 因果注意力矩阵\n(下三角模式)')
    plt.xticks(range(seq_len), [f'Pos {i}' for i in range(seq_len)])
    plt.yticks(range(seq_len), [f'Pos {i}' for i in range(seq_len)])
    plt.grid(False)
    
    # 添加解释
    plt.figtext(0.5, 0.01, 
               '● = 允许关注  |  · = 禁止关注\n每个位置只能关注当前位置及之前的位置', 
               ha='center', fontsize=12, style='italic')
    
    plt.tight_layout()
    plt.show()

# 绘制注意力矩阵
plot_causal_attention_matrix(4)