构建AI智能体：信息论完全指南：从基础概念到在大模型中的实际应用

# 环境设置和库导入
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from collections import Counter
import math
from scipy.special import entr
import warnings
warnings.filterwarnings('ignore')

# 设置中文字体（如果遇到显示问题可以注释掉）
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False
def information_content(probability):
    """
    计算单个事件的信息量
    
    Parameters:
    probability: 事件发生的概率，范围[0,1]
    
    Returns:
    information: 信息量（比特）
    """
    if probability == 0:
        return float('inf')  # 不可能事件的信息量为无穷大
    return -math.log2(probability)

# 示例计算
print("=== 信息量计算示例 ===")
events = [
    ("太阳从东边升起", 0.999),
    ("今天会下雨", 0.3),
    ("赢得彩票", 0.0000001),
    ("抛硬币正面朝上", 0.5)
]

for event_name, prob in events:
    info = information_content(prob)
    print(f"事件: {event_name:20} 概率: {prob:8.6f} 信息量: {info:8.2f} bits")

# 可视化信息量与概率的关系
probabilities = np.linspace(0.001, 1.0, 1000)
information_values = [information_content(p) for p in probabilities]

plt.figure(figsize=(10, 6))
plt.plot(probabilities, information_values, 'b-', linewidth=2, alpha=0.8)
plt.xlabel('概率 P(x)')
plt.ylabel('信息量 I(x) (bits)')
plt.title('信息量与概率的关系\nI(x) = -log₂P(x)')
plt.grid(True, alpha=0.3)

# 标记关键点
key_points = [0.1, 0.5, 0.9]
for p in key_points:
    info = information_content(p)
    plt.plot(p, info, 'ro', markersize=8)
    plt.annotate(f'P={p}, I={info:.2f}', (p, info), 
                xytext=(10, 10), textcoords='offset points',
                bbox=dict(boxstyle='round,pad=0.3', facecolor='yellow', alpha=0.7))

plt.tight_layout()
plt.show()

# 环境设置和库导入
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from collections import Counter
import math
from scipy.special import entr
import warnings
warnings.filterwarnings('ignore')

# 设置中文字体（如果遇到显示问题可以注释掉）
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

def entropy(probabilities):
    """
    计算概率分布的信息熵
    
    Parameters:
    probabilities: 概率列表，应该和为1
    
    Returns:
    entropy_value: 信息熵值（比特）
    """
    # 输入验证
    probabilities = np.array(probabilities)
    if not np.allclose(np.sum(probabilities), 1.0):
        raise ValueError("概率之和必须为1")
    if np.any(probabilities < 0):
        raise ValueError("概率不能为负")
    
    entropy_val = 0.0
    for p in probabilities:
        if p > 0:  # 避免log(0)的情况
            entropy_val -= p * math.log2(p)
    return entropy_val

def entropy_from_labels(labels):
    """
    从标签数据直接计算熵
    
    Parameters:
    labels: 类别标签数组
    
    Returns:
    entropy_value: 信息熵值
    """
    counter = Counter(labels)
    total = len(labels)
    probabilities = [count / total for count in counter.values()]
    return entropy(probabilities)

print("=== 信息熵计算示例 ===")

# 不同的概率分布示例
distributions = {
    "完全确定": [1.0],
    "高度倾斜": [0.9, 0.1],
    "中等倾斜": [0.7, 0.3],
    "公平硬币": [0.5, 0.5],
    "公平骰子": [1/6, 1/6, 1/6, 1/6, 1/6, 1/6],
    "三类别均匀": [1/3, 1/3, 1/3]
}

print(f"{'分布类型':<15} {'概率分布':<40} {'熵值':<8} {'最大可能熵':<12}")
print("-" * 85)

for dist_name, probs in distributions.items():
    h = entropy(probs)
    max_possible = math.log2(len(probs)) if len(probs) > 0 else 0
    efficiency = (h / max_possible * 100) if max_possible > 0 else 0
    
    prob_str = str([round(p, 3) for p in probs])
    print(f"{dist_name:<15} {prob_str:<40} {h:<8.4f} {max_possible:<12.4f}")

# 二分类熵曲线可视化
p_values = np.linspace(0.001, 0.999, 1000)
binary_entropies = [entropy([p, 1-p]) for p in p_values]

plt.figure(figsize=(12, 6))
plt.plot(p_values, binary_entropies, 'r-', linewidth=3, alpha=0.8, label='二分类熵')
plt.axvline(x=0.5, color='blue', linestyle='--', alpha=0.7, 
           label='最大熵点 (p=0.5)')
plt.axhline(y=1.0, color='green', linestyle='--', alpha=0.7, 
           label='最大熵值 (1 bit)')

plt.xlabel('类别1的概率 (p)')
plt.ylabel('熵 H(X) (bits)')
plt.title('二分类信息熵曲线\nH(X) = -[p·log₂(p) + (1-p)·log₂(1-p)]')
plt.legend()
plt.grid(True, alpha=0.3)

# 添加一些关键点的标注
key_probabilities = [0.1, 0.25, 0.5, 0.75, 0.9]
for p in key_probabilities:
    h_val = entropy([p, 1-p])
    plt.plot(p, h_val, 'ko', markersize=6)
    plt.annotate(f'p={p}\nH={h_val:.3f}', (p, h_val),
                xytext=(20, 20), textcoords='offset points',
                arrowprops=dict(arrowstyle='->', connectionstyle='arc3,rad=0'),
                bbox=dict(boxstyle='round,pad=0.3', facecolor='lightblue', alpha=0.7))

plt.tight_layout()
plt.show()

# 环境设置和库导入
import numpy as np
import matplotlib.pyplot as plt
import math
from scipy.stats import entropy
import warnings
warnings.filterwarnings('ignore')

# 设置中文字体（如果遇到显示问题可以注释掉）
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

def model_uncertainty_analysis(predictions):
    """
    分析模型预测的不确定性
    
    Parameters:
    predictions: 模型输出的概率分布列表
    
    Returns:
    average_entropy: 平均熵值
    uncertainty_ratio: 不确定性比例
    """
    entropies = [entropy(pred) for pred in predictions]
    avg_entropy = np.mean(entropies)
    max_entropy = math.log2(len(predictions[0]))  # 假设所有预测维度相同
    uncertainty_ratio = avg_entropy / max_entropy
    
    return avg_entropy, uncertainty_ratio

# 模拟模型预测
sample_predictions = [
    [0.9, 0.1],    # 确定性高的预测
    [0.6, 0.4],    # 中等确定性
    [0.5, 0.5],    # 完全不确定
    [0.8, 0.2],    # 较高确定性
    [0.7, 0.3]     # 中等确定性
]

avg_ent, uncertainty = model_uncertainty_analysis(sample_predictions)
print(f"模型预测分析:")
print(f"  平均熵值: {avg_ent:.4f} bits")
print(f"  不确定性比例: {uncertainty:.2%}")

# 环境设置和库导入
import numpy as np
import matplotlib.pyplot as plt
import math
from scipy.stats import entropy
from collections import Counter
import warnings
warnings.filterwarnings('ignore')

def text_complexity_analysis(text):
    """
    分析文本的复杂度（基于字符级熵）
    
    Parameters:
    text: 输入文本
    
    Returns:
    complexity: 文本复杂度（熵值）
    """
    char_counts = Counter(text)
    total_chars = len(text)
    probabilities = [count/total_chars for count in char_counts.values()]
    return entropy(probabilities)

# 示例文本分析
sample_texts = [
    "aaaaabbbbbcccccdddddeeeee",  # 低复杂度
    "hello world this is a test",  # 中等复杂度
    "the quick brown fox jumps over the lazy dog",  # 较高复杂度
    "abc123def456ghi789jkl0",     # 高复杂度
]

print("\n=== 文本复杂度分析 ===")
for text in sample_texts:
    comp = text_complexity_analysis(text)
    print(f"文本: {text[:30]:<30} 复杂度: {comp:.4f} bits/字符")

# 环境设置和库导入
import numpy as np
import matplotlib.pyplot as plt
import math
from scipy.stats import entropy
from collections import Counter
import warnings
warnings.filterwarnings('ignore')

# 设置中文字体（如果遇到显示问题可以注释掉）
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

def joint_entropy(joint_prob_matrix):
    """
    计算两个随机变量的联合熵
    
    Parameters:
    joint_prob_matrix: 联合概率矩阵，P(X,Y)
    
    Returns:
    joint_entropy_value: 联合熵值
    """
    joint_prob_matrix = np.array(joint_prob_matrix)
    
    # 验证输入
    if not np.allclose(np.sum(joint_prob_matrix), 1.0):
        raise ValueError("联合概率矩阵之和必须为1")
    if np.any(joint_prob_matrix < 0):
        raise ValueError("概率不能为负")
    
    entropy_val = 0.0
    for i in range(joint_prob_matrix.shape[0]):
        for j in range(joint_prob_matrix.shape[1]):
            p = joint_prob_matrix[i, j]
            if p > 0:
                entropy_val -= p * math.log2(p)
    return entropy_val

print("=== 联合熵计算示例 ===")

# 示例1：天气和活动的联合分布
# 行：天气（0=晴天，1=雨天），列：活动（0=室内，1=室外）
print("示例1：天气和活动的联合分布")

# 情况A：独立变量
joint_independent = np.array([
    [0.3, 0.2],  # 晴天：室内0.3，室外0.2
    [0.3, 0.2]   # 雨天：室内0.3，室外0.2
])

# 情况B：依赖变量
joint_dependent = np.array([
    [0.4, 0.1],  # 晴天：室内0.4，室外0.1
    [0.1, 0.4]   # 雨天：室内0.1，室外0.4
])

cases = [("独立变量", joint_independent), ("依赖变量", joint_dependent)]

for case_name, joint_probs in cases:
    h_joint = joint_entropy(joint_probs)
    h_weather = entropy(np.sum(joint_probs, axis=1))  # 天气的边际熵
    h_activity = entropy(np.sum(joint_probs, axis=0)) # 活动的边际熵
    
    print(f"\n{case_name}:")
    print(f"  联合概率分布:")
    print(f"       活动=0   活动=1")
    print(f"  天气=0  {joint_probs[0,0]:.2f}     {joint_probs[0,1]:.2f}")
    print(f"  天气=1  {joint_probs[1,0]:.2f}     {joint_probs[1,1]:.2f}")
    print(f"  联合熵 H(天气,活动): {h_joint:.4f}")
    print(f"  天气熵 H(天气): {h_weather:.4f}")
    print(f"  活动熵 H(活动): {h_activity:.4f}")
    print(f"  边际熵之和: {h_weather + h_activity:.4f}")
    print(f"  关系验证: H(X,Y) ≤ H(X) + H(Y): {h_joint <= h_weather + h_activity}")

# 可视化联合分布
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 6))

# 独立情况
im1 = ax1.imshow(joint_independent, cmap='Blues', aspect='auto')
ax1.set_xticks([0, 1])
ax1.set_yticks([0, 1])
ax1.set_xticklabels(['室内', '室外'])
ax1.set_yticklabels(['晴天', '雨天'])
ax1.set_title('独立变量联合分布\nH(X,Y) = H(X) + H(Y)')
plt.colorbar(im1, ax=ax1)

# 添加数值标注
for i in range(2):
    for j in range(2):
        ax1.text(j, i, f'{joint_independent[i, j]:.2f}', 
                ha='center', va='center', fontsize=12, color='red')

# 依赖情况
im2 = ax2.imshow(joint_dependent, cmap='Blues', aspect='auto')
ax2.set_xticks([0, 1])
ax2.set_yticks([0, 1])
ax2.set_xticklabels(['室内', '室外'])
ax2.set_yticklabels(['晴天', '雨天'])
ax2.set_title('依赖变量联合分布\nH(X,Y) < H(X) + H(Y)')
plt.colorbar(im2, ax=ax2)

# 添加数值标注
for i in range(2):
    for j in range(2):
        ax2.text(j, i, f'{joint_dependent[i, j]:.2f}', 
                ha='center', va='center', fontsize=12, color='red')

plt.tight_layout()
plt.show()

# 环境设置和库导入
import numpy as np
import matplotlib.pyplot as plt
import math
from scipy.stats import entropy
from collections import Counter
import warnings
warnings.filterwarnings('ignore')

# 设置中文字体（如果遇到显示问题可以注释掉）
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

def joint_entropy(joint_prob_matrix):
    """
    计算两个随机变量的联合熵
    
    Parameters:
    joint_prob_matrix: 联合概率矩阵，P(X,Y)
    
    Returns:
    joint_entropy_value: 联合熵值
    """
    joint_prob_matrix = np.array(joint_prob_matrix)
    
    # 验证输入
    if not np.allclose(np.sum(joint_prob_matrix), 1.0):
        raise ValueError("联合概率矩阵之和必须为1")
    if np.any(joint_prob_matrix < 0):
        raise ValueError("概率不能为负")
    
    entropy_val = 0.0
    for i in range(joint_prob_matrix.shape[0]):
        for j in range(joint_prob_matrix.shape[1]):
            p = joint_prob_matrix[i, j]
            if p > 0:
                entropy_val -= p * math.log2(p)
    return entropy_val

def conditional_entropy(joint_probs):
    """
    计算条件熵 H(Y|X)
    
    Parameters:
    joint_probs: 联合概率矩阵 P(X,Y)
    
    Returns:
    conditional_entropy_value: 条件熵值
    """
    joint_probs = np.array(joint_probs)
    
    # 计算边际分布 P(X)
    p_x = np.sum(joint_probs, axis=1)
    
    cond_entropy = 0.0
    for i in range(joint_probs.shape[0]):
        if p_x[i] > 0:
            # 计算条件分布 P(Y|X=x_i)
            p_y_given_x = joint_probs[i, :] / p_x[i]
            # 计算条件分布的熵
            h_y_given_x = entropy(p_y_given_x)
            # 加权平均
            cond_entropy += p_x[i] * h_y_given_x
    
    return cond_entropy

print("=== 条件熵计算示例 ===")

# 使用之前的依赖变量例子
joint_probs = np.array([
    [0.4, 0.1],  # 晴天：室内0.4，室外0.1
    [0.1, 0.4]   # 雨天：室内0.1，室外0.4
])

h_activity_given_weather = conditional_entropy(joint_probs)
h_joint = joint_entropy(joint_probs)
h_weather = entropy(np.sum(joint_probs, axis=1))
h_activity = entropy(np.sum(joint_probs, axis=0))

print("依赖变量案例详细分析:")
print(f"  天气熵 H(Weather): {h_weather:.4f}")
print(f"  活动熵 H(Activity): {h_activity:.4f}")
print(f"  联合熵 H(Weather, Activity): {h_joint:.4f}")
print(f"  条件熵 H(Activity|Weather): {h_activity_given_weather:.4f}")

# 验证链式法则
print(f"\n链式法则验证:")
print(f"  H(Weather) + H(Activity|Weather) = {h_weather:.4f} + {h_activity_given_weather:.4f} = {h_weather + h_activity_given_weather:.4f}")
print(f"  H(Weather, Activity) = {h_joint:.4f}")
print(f"  差值: {abs((h_weather + h_activity_given_weather) - h_joint):.8f}")

# 与独立情况对比
print(f"\n与独立情况对比:")
joint_indep = np.array([
    [0.25, 0.25],
    [0.25, 0.25]
])
h_activity_given_weather_indep = conditional_entropy(joint_indep)
print(f"  独立情况 - H(Activity|Weather): {h_activity_given_weather_indep:.4f}")
print(f"  依赖情况 - H(Activity|Weather): {h_activity_given_weather:.4f}")
print(f"  知道天气在依赖情况下减少更多不确定性")

# 不确定性减少比例
reduction_dep = h_activity - h_activity_given_weather
reduction_indep = h_activity - h_activity_given_weather_indep
print(f"\n不确定性减少量:")
print(f"  依赖情况减少: {reduction_dep:.4f} bits ({reduction_dep/h_activity*100:.1f}%)")
print(f"  独立情况减少: {reduction_indep:.4f} bits ({reduction_indep/h_activity*100:.1f}%)")

# 环境设置和库导入
import numpy as np
import matplotlib.pyplot as plt
import math
from scipy.stats import entropy
from collections import Counter
import warnings
warnings.filterwarnings('ignore')

# 设置中文字体（如果遇到显示问题可以注释掉）
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

def joint_entropy(joint_prob_matrix):
    """
    计算两个随机变量的联合熵
    
    Parameters:
    joint_prob_matrix: 联合概率矩阵，P(X,Y)
    
    Returns:
    joint_entropy_value: 联合熵值
    """
    joint_prob_matrix = np.array(joint_prob_matrix)
    
    # 验证输入
    if not np.allclose(np.sum(joint_prob_matrix), 1.0):
        raise ValueError("联合概率矩阵之和必须为1")
    if np.any(joint_prob_matrix < 0):
        raise ValueError("概率不能为负")
    
    entropy_val = 0.0
    for i in range(joint_prob_matrix.shape[0]):
        for j in range(joint_prob_matrix.shape[1]):
            p = joint_prob_matrix[i, j]
            if p > 0:
                entropy_val -= p * math.log2(p)
    return entropy_val

def mutual_information(joint_probs):
    """
    计算两个随机变量的互信息
    
    Parameters:
    joint_probs: 联合概率矩阵 P(X,Y)
    
    Returns:
    mi: 互信息值
    mi_alt: 另一种计算方法的结果（用于验证）
    """
    joint_probs = np.array(joint_probs)
    
    # 方法1：使用熵的定义
    p_x = np.sum(joint_probs, axis=1)  # P(X)
    p_y = np.sum(joint_probs, axis=0)  # P(Y)
    
    h_x = entropy(p_x)
    h_y = entropy(p_y)
    h_xy = joint_entropy(joint_probs)
    
    mi = h_x + h_y - h_xy
    
    # 方法2：直接计算
    mi_alt = 0.0
    for i in range(joint_probs.shape[0]):
        for j in range(joint_probs.shape[1]):
            p_xy = joint_probs[i, j]
            if p_xy > 0:
                mi_alt += p_xy * math.log2(p_xy / (p_x[i] * p_y[j]))
    
    return mi, mi_alt

print("=== 互信息计算示例 ===")

# 不同依赖程度的案例
cases = [
    ("强依赖", np.array([[0.4, 0.1], [0.1, 0.4]])),
    ("中等依赖", np.array([[0.35, 0.15], [0.15, 0.35]])),
    ("弱依赖", np.array([[0.3, 0.2], [0.2, 0.3]])),
    ("独立", np.array([[0.25, 0.25], [0.25, 0.25]]))
]

print(f"{'案例':<10} {'H(X)':<8} {'H(Y)':<8} {'H(X,Y)':<8} {'I(X;Y)':<8} {'归一化MI':<10}")
print("-" * 65)

for case_name, joint_probs in cases:
    mi, mi_alt = mutual_information(joint_probs)
    h_x = entropy(np.sum(joint_probs, axis=1))
    h_y = entropy(np.sum(joint_probs, axis=0))
    h_xy = joint_entropy(joint_probs)
    
    # 归一化互信息（0到1之间）
    normalized_mi = mi / min(h_x, h_y) if min(h_x, h_y) > 0 else 0
    
    print(f"{case_name:<10} {h_x:<8.4f} {h_y:<8.4f} {h_xy:<8.4f} {mi:<8.4f} {normalized_mi:<10.4f}")

# 验证两种计算方法的一致性
print(f"\n计算方法验证:")
for case_name, joint_probs in cases[:2]:  # 只验证前两个案例
    mi, mi_alt = mutual_information(joint_probs)
    print(f"{case_name}: 方法1={mi:.6f}, 方法2={mi_alt:.6f}, 差值={abs(mi-mi_alt):.8f}")

# 互信息可视化
fig, axes = plt.subplots(2, 2, figsize=(12, 10))
axes = axes.ravel()

for idx, (case_name, joint_probs) in enumerate(cases):
    mi, _ = mutual_information(joint_probs)
    h_x = entropy(np.sum(joint_probs, axis=1))
    h_y = entropy(np.sum(joint_probs, axis=0))
    
    # 绘制韦恩图风格的表示
    circle1 = plt.Circle((0.3, 0.5), 0.2, color='blue', alpha=0.3, label='H(X)')
    circle2 = plt.Circle((0.7, 0.5), 0.2, color='red', alpha=0.3, label='H(Y)')
    
    axes[idx].add_patch(circle1)
    axes[idx].add_patch(circle2)
    
    # 设置图形属性
    axes[idx].set_xlim(0, 1)
    axes[idx].set_ylim(0, 1)
    axes[idx].set_aspect('equal')
    axes[idx].set_title(f'{case_name}\nI(X;Y) = {mi:.4f}')
    axes[idx].text(0.5, 0.9, f'H(X)={h_x:.3f}', ha='center', fontsize=10)
    axes[idx].text(0.5, 0.1, f'H(Y)={h_y:.3f}', ha='center', fontsize=10)
    axes[idx].text(0.5, 0.5, f'MI={mi:.3f}', ha='center', fontsize=12, 
                  bbox=dict(boxstyle='round', facecolor='yellow', alpha=0.7))
    
    # 隐藏坐标轴
    axes[idx].set_xticks([])
    axes[idx].set_yticks([])

plt.tight_layout()
plt.show()

# 环境设置和库导入
import numpy as np
import math
from scipy.stats import entropy
import warnings
warnings.filterwarnings('ignore')

def kl_divergence(p, q):
    """
    计算KL散度 D_KL(P || Q)
    
    Parameters:
    p: 真实分布P
    q: 近似分布Q
    
    Returns:
    kl_divergence_value: KL散度值
    """
    p = np.array(p)
    q = np.array(q)
    
    # 输入验证
    if len(p) != len(q):
        raise ValueError("分布P和Q必须有相同的长度")
    if not np.allclose(np.sum(p), 1.0) or not np.allclose(np.sum(q), 1.0):
        raise ValueError("分布P和Q必须和为1")
    if np.any(p < 0) or np.any(q < 0):
        raise ValueError("概率不能为负")
    
    divergence = 0.0
    for i in range(len(p)):
        if p[i] > 0:
            if q[i] > 0:
                divergence += p[i] * math.log2(p[i] / q[i])
            else:
                return float('inf')  # 如果Q在P有概率的地方为0，散度为无穷大
    return divergence

print("=== KL散度计算示例 ===")

# 真实分布（例如真实的词频分布）
p_true = np.array([0.5, 0.3, 0.15, 0.05])

# 不同的模型近似
approximations = [
    ("优秀近似", np.array([0.48, 0.31, 0.14, 0.07])),
    ("中等近似", np.array([0.6, 0.25, 0.1, 0.05])),
    ("较差近似", np.array([0.7, 0.2, 0.08, 0.02])),
    ("错误近似", np.array([0.1, 0.1, 0.4, 0.4])),
    ("零概率问题", np.array([0.5, 0.3, 0.2, 0.0]))  # 在P>0的地方Q=0
]

print(f"真实分布 P: {p_true}")
print("\n不同近似的KL散度:")
print(f"{'近似类型':<12} {'分布Q':<30} {'D_KL(P||Q)':<12} {'评估':<10}")
print("-" * 75)

for name, q in approximations:
    try:
        kl = kl_divergence(p_true, q)
        if kl < 0.1:
            assessment = "优秀"
        elif kl < 0.5:
            assessment = "良好"
        elif kl < 1.0:
            assessment = "一般"
        else:
            assessment = "较差"
    except:
        kl = float('inf')
        assessment = "无效"
    
    q_str = str([round(val, 3) for val in q])
    print(f"{name:<12} {q_str:<30} {kl:<12.4f} {assessment:<10}")

# 不对称性演示
print(f"\n=== KL散度不对称性演示 ===")
p = np.array([0.7, 0.3])
q = np.array([0.4, 0.6])

kl_pq = kl_divergence(p, q)
kl_qp = kl_divergence(q, p)

print(f"分布 P: {p}")
print(f"分布 Q: {q}")
print(f"D_KL(P||Q) = {kl_pq:.4f}")
print(f"D_KL(Q||P) = {kl_qp:.4f}")
print(f"不对称性: D_KL(P||Q) ≠ D_KL(Q||P)")

# KL散度与模型训练的关系
print(f"\n=== KL散度在模型训练中的应用 ===")
print("在机器学习中，我们通常最小化 D_KL(P_data || P_model)")
print("这等价于最大似然估计")
print("KL散度作为正则项可以防止模型过度偏离先验分布")

# 环境设置和库导入
import numpy as np
import math
from scipy.stats import entropy
import warnings
warnings.filterwarnings('ignore')

def kl_divergence(p, q):
    """
    计算KL散度 D_KL(P || Q)
    
    Parameters:
    p: 真实分布P
    q: 近似分布Q
    
    Returns:
    kl_divergence_value: KL散度值
    """
    p = np.array(p)
    q = np.array(q)
    
    # 输入验证
    if len(p) != len(q):
        raise ValueError("分布P和Q必须有相同的长度")
    if not np.allclose(np.sum(p), 1.0) or not np.allclose(np.sum(q), 1.0):
        raise ValueError("分布P和Q必须和为1")
    if np.any(p < 0) or np.any(q < 0):
        raise ValueError("概率不能为负")
    
    divergence = 0.0
    for i in range(len(p)):
        if p[i] > 0:
            if q[i] > 0:
                divergence += p[i] * math.log2(p[i] / q[i])
            else:
                return float('inf')  # 如果Q在P有概率的地方为0，散度为无穷大
    return divergence

def cross_entropy(p, q):
    """
    计算交叉熵 H(P, Q)
    
    Parameters:
    p: 真实分布P
    q: 预测分布Q
    
    Returns:
    cross_entropy_value: 交叉熵值
    """
    p = np.array(p)
    q = np.array(q)
    
    # 输入验证
    if len(p) != len(q):
        raise ValueError("分布P和Q必须有相同的长度")
    if not np.allclose(np.sum(p), 1.0) or not np.allclose(np.sum(q), 1.0):
        raise ValueError("分布P和Q必须和为1")
    if np.any(p < 0) or np.any(q < 0):
        raise ValueError("概率不能为负")
    
    ce = 0.0
    for i in range(len(p)):
        if p[i] > 0:
            if q[i] > 0:
                ce -= p[i] * math.log2(q[i])
            else:
                return float('inf')  # 如果Q在P有概率的地方为0，交叉熵为无穷大
    return ce

print("=== 交叉熵计算示例 ===")

# 真实分布和不同的预测分布
p_true = np.array([0.6, 0.3, 0.1])

predictions = [
    ("完美预测", np.array([0.6, 0.3, 0.1])),
    ("良好预测", np.array([0.5, 0.4, 0.1])),
    ("一般预测", np.array([0.7, 0.2, 0.1])),
    ("较差预测", np.array([0.8, 0.1, 0.1])),
    ("错误预测", np.array([0.1, 0.1, 0.8]))
]

print(f"真实分布 P: {p_true}")
print(f"{'预测类型':<12} {'预测分布Q':<25} {'交叉熵':<10} {'KL散度':<10} {'真实熵':<10}")
print("-" * 80)

for name, q in predictions:
    ce = cross_entropy(p_true, q)
    h_p = entropy(p_true)
    kl = kl_divergence(p_true, q)
    
    print(f"{name:<12} {str([round(val, 3) for val in q]):<25} {ce:<10.4f} {kl:<10.4f} {h_p:<10.4f}")

# 验证交叉熵与KL散度的关系
print(f"\n=== 交叉熵与KL散度关系验证 ===")
p_test = np.array([0.5, 0.3, 0.2])
q_test = np.array([0.4, 0.4, 0.2])

ce_val = cross_entropy(p_test, q_test)
h_p_val = entropy(p_test)
kl_val = kl_divergence(p_test, q_test)

print(f"真实分布 P: {p_test}")
print(f"预测分布 Q: {q_test}")
print(f"交叉熵 H(P,Q): {ce_val:.4f}")
print(f"真实熵 H(P): {h_p_val:.4f}")
print(f"KL散度 D_KL(P||Q): {kl_val:.4f}")
print(f"关系验证: H(P) + D_KL(P||Q) = {h_p_val:.4f} + {kl_val:.4f} = {h_p_val + kl_val:.4f}")
print(f"与交叉熵相等: {abs((h_p_val + kl_val) - ce_val) < 1e-10}")

# 二分类交叉熵示例
print(f"\n=== 二分类交叉熵示例 ===")
def binary_cross_entropy(y_true, y_pred):
    """
    计算二分类交叉熵
    
    Parameters:
    y_true: 真实标签 (0或1)
    y_pred: 预测概率 (0到1之间)
    
    Returns:
    bce: 平均交叉熵
    """
    y_true = np.array(y_true)
    y_pred = np.array(y_pred)
    
    # 避免log(0)的情况
    y_pred = np.clip(y_pred, 1e-15, 1 - 1e-15)
    
    bce = -np.mean(y_true * np.log2(y_pred) + (1 - y_true) * np.log2(1 - y_pred))
    return bce

# 示例数据
y_true = np.array([1, 0, 1, 1, 0])
y_pred_good = np.array([0.9, 0.2, 0.8, 0.7, 0.3])
y_pred_poor = np.array([0.6, 0.5, 0.6, 0.6, 0.4])

bce_good = binary_cross_entropy(y_true, y_pred_good)
bce_poor = binary_cross_entropy(y_true, y_pred_poor)

print(f"真实标签: {y_true}")
print(f"良好预测: {y_pred_good} → 交叉熵: {bce_good:.4f}")
print(f"较差预测: {y_pred_poor} → 交叉熵: {bce_poor:.4f}")

# 环境设置和库导入
import numpy as np
import math
from scipy.stats import entropy
import warnings
warnings.filterwarnings('ignore')

def kl_divergence(p, q):
    """
    计算KL散度 D_KL(P || Q)
    
    Parameters:
    p: 真实分布P
    q: 近似分布Q
    
    Returns:
    kl_divergence_value: KL散度值
    """
    p = np.array(p)
    q = np.array(q)
    
    # 输入验证
    if len(p) != len(q):
        raise ValueError("分布P和Q必须有相同的长度")
    if not np.allclose(np.sum(p), 1.0) or not np.allclose(np.sum(q), 1.0):
        raise ValueError("分布P和Q必须和为1")
    if np.any(p < 0) or np.any(q < 0):
        raise ValueError("概率不能为负")
    
    divergence = 0.0
    for i in range(len(p)):
        if p[i] > 0:
            if q[i] > 0:
                divergence += p[i] * math.log2(p[i] / q[i])
            else:
                return float('inf')  # 如果Q在P有概率的地方为0，散度为无穷大
    return divergence

def cross_entropy(p, q):
    """
    计算交叉熵 H(P, Q)
    
    Parameters:
    p: 真实分布P
    q: 预测分布Q
    
    Returns:
    cross_entropy_value: 交叉熵值
    """
    p = np.array(p)
    q = np.array(q)
    
    # 输入验证
    if len(p) != len(q):
        raise ValueError("分布P和Q必须有相同的长度")
    if not np.allclose(np.sum(p), 1.0) or not np.allclose(np.sum(q), 1.0):
        raise ValueError("分布P和Q必须和为1")
    if np.any(p < 0) or np.any(q < 0):
        raise ValueError("概率不能为负")
    
    ce = 0.0
    for i in range(len(p)):
        if p[i] > 0:
            if q[i] > 0:
                ce -= p[i] * math.log2(q[i])
            else:
                return float('inf')  # 如果Q在P有概率的地方为0，交叉熵为无穷大
    return ce

class LanguageModelTrainer:
    """简化的语言模型训练器，演示交叉熵的应用"""
    
    def __init__(self, vocab_size=1000):
        self.vocab_size = vocab_size
        
    def calculate_batch_loss(self, true_distributions, predicted_distributions):
        """
        计算批量的交叉熵损失
        
        Parameters:
        true_distributions: 真实分布列表
        predicted_distributions: 预测分布列表
        
        Returns:
        average_loss: 平均交叉熵损失
        loss_breakdown: 每个样本的损失详情
        """
        batch_size = len(true_distributions)
        losses = []
        
        for i in range(batch_size):
            ce_loss = cross_entropy(true_distributions[i], predicted_distributions[i])
            losses.append(ce_loss)
        
        average_loss = np.mean(losses)
        
        loss_breakdown = {
            'average_loss': average_loss,
            'min_loss': np.min(losses),
            'max_loss': np.max(losses),
            'std_loss': np.std(losses),
            'losses': losses
        }
        
        return average_loss, loss_breakdown
    
    def analyze_training_progress(self, epoch_losses):
        """
        分析训练进度
        
        Parameters:
        epoch_losses: 每个epoch的损失列表
        
        Returns:
        analysis: 训练分析结果
        """
        analysis = {
            'final_loss': epoch_losses[-1],
            'best_loss': np.min(epoch_losses),
            'improvement': epoch_losses[0] - epoch_losses[-1],
            'convergence_rate': self._calculate_convergence_rate(epoch_losses)
        }
        return analysis
    
    def _calculate_convergence_rate(self, losses):
        """计算收敛速率"""
        if len(losses) < 2:
            return 0
        improvements = []
        for i in range(1, len(losses)):
            improvement = losses[i-1] - losses[i]
            improvements.append(improvement)
        return np.mean(improvements)

# 模拟训练过程
print("=== 语言模型训练模拟 ===")
trainer = LanguageModelTrainer(vocab_size=1000)

# 模拟训练数据（真实分布和预测分布）
batch_size = 5
vocab_size = 10

# 生成模拟数据
true_dists = []
pred_dists = []

for i in range(batch_size):
    # 真实分布（one-hot或接近one-hot）
    true_dist = np.zeros(vocab_size)
    true_class = np.random.randint(0, vocab_size)
    true_dist[true_class] = 1.0
    true_dists.append(true_dist)
    
    # 预测分布（模型输出）
    pred_dist = np.random.dirichlet(np.ones(vocab_size) * 0.1)  # 狄利克雷分布
    pred_dists.append(pred_dist)

# 计算损失
avg_loss, loss_details = trainer.calculate_batch_loss(true_dists, pred_dists)

print(f"批量训练结果:")
print(f"  平均损失: {avg_loss:.4f} bits")
print(f"  最小损失: {loss_details['min_loss']:.4f} bits")
print(f"  最大损失: {loss_details['max_loss']:.4f} bits")
print(f"  损失标准差: {loss_details['std_loss']:.4f} bits")

# 模拟训练过程
print(f"\n=== 训练过程模拟 ===")
epochs = 20
simulated_losses = []

# 模拟损失下降（指数衰减）
initial_loss = 8.0
for epoch in range(epochs):
    # 模拟损失下降
    loss = initial_loss * np.exp(-epoch / 5) + np.random.normal(0, 0.1)
    simulated_losses.append(loss)
    print(f"Epoch {epoch+1:2d}: 损失 = {loss:.4f} bits")

# 分析训练进度
analysis = trainer.analyze_training_progress(simulated_losses)
print(f"\n训练分析:")
print(f"  最终损失: {analysis['final_loss']:.4f} bits")
print(f"  最佳损失: {analysis['best_loss']:.4f} bits")
print(f"  总改进: {analysis['improvement']:.4f} bits")
print(f"  平均每epoch改进: {analysis['convergence_rate']:.4f} bits")

# 环境设置和库导入
import numpy as np
import math
from scipy.stats import entropy
from collections import Counter
import warnings
warnings.filterwarnings('ignore')

def entropy_from_labels(labels):
    """
    从标签数据直接计算熵
    
    Parameters:
    labels: 类别标签数组
    
    Returns:
    entropy_value: 信息熵值
    """
    counter = Counter(labels)
    total = len(labels)
    probabilities = [count / total for count in counter.values()]
    return entropy(probabilities)

def information_gain(feature_values, labels):
    """
    计算特征的信息增益
    
    Parameters:
    feature_values: 特征值数组
    labels: 目标标签数组
    
    Returns:
    information_gain_value: 信息增益值
    split_info: 分割的详细信息
    """
    # 原始熵 H(Y)
    original_entropy = entropy_from_labels(labels)
    
    # 条件熵 H(Y|X)
    unique_features = np.unique(feature_values)
    conditional_entropy_val = 0.0
    
    split_info = {
        'original_entropy': original_entropy,
        'feature_values': unique_features,
        'splits': []
    }
    
    for feature_val in unique_features:
        mask = feature_values == feature_val
        subset_labels = labels[mask]
        
        if len(subset_labels) > 0:
            weight = len(subset_labels) / len(labels)
            subset_entropy = entropy_from_labels(subset_labels)
            conditional_entropy_val += weight * subset_entropy
            
            # 记录分割信息
            split_info['splits'].append({
                'feature_value': feature_val,
                'weight': weight,
                'subset_entropy': subset_entropy,
                'subset_size': len(subset_labels),
                'label_distribution': dict(Counter(subset_labels))
            })
    
    ig = original_entropy - conditional_entropy_val
    split_info['conditional_entropy'] = conditional_entropy_val
    split_info['information_gain'] = ig
    
    return ig, split_info

print("=== 信息增益计算示例 ===")

# 学生成绩预测数据集
# 特征：学习时间 (0=少, 1=多)
# 目标：成绩 (0=不及格, 1=及格)

study_time = np.array([0, 0, 0, 0, 1, 1, 1, 1])
grades = np.array([0, 0, 1, 0, 1, 1, 1, 1])

print("数据集:")
print(f"学习时间: {study_time} (0=少, 1=多)")
print(f"成绩:     {grades} (0=不及格, 1=及格)")

# 计算信息增益
ig, split_info = information_gain(study_time, grades)

print(f"\n信息增益分析:")
print(f"原始熵 H(成绩): {split_info['original_entropy']:.4f} bits")
print(f"条件熵 H(成绩|学习时间): {split_info['conditional_entropy']:.4f} bits")
print(f"信息增益 IG(成绩, 学习时间): {ig:.4f} bits")
print(f"不确定性减少比例: {ig/split_info['original_entropy']*100:.1f}%")

# 详细分割信息
print(f"\n详细分割信息:")
for split in split_info['splits']:
    study_desc = "学习时间少" if split['feature_value'] == 0 else "学习时间多"
    print(f"  {study_desc}:")
    print(f"    权重: {split['weight']:.2f}")
    print(f"    子集大小: {split['subset_size']}")
    print(f"    子集熵: {split['subset_entropy']:.4f}")
    print(f"    标签分布: {split['label_distribution']}")

# 多特征比较
print(f"\n=== 多特征信息增益比较 ===")

# 添加更多特征
temperature = np.array([1, 1, 0, 0, 0, 0, 1, 1])  # 0=冷, 1=暖
random_feature = np.array([0, 1, 0, 1, 0, 1, 0, 1])  # 随机特征

features = [
    ("学习时间", study_time),
    ("温度", temperature),
    ("随机特征", random_feature)
]

print("特征比较:")
print(f"{'特征':<10} {'信息增益':<12} {'归一化增益':<12} {'评估':<10}")
print("-" * 50)

for feature_name, feature_vals in features:
    ig_val, _ = information_gain(feature_vals, grades)
    normalized_ig = ig_val / entropy_from_labels(grades)
    
    if normalized_ig > 0.7:
        assessment = "优秀"
    elif normalized_ig > 0.3:
        assessment = "良好"
    elif normalized_ig > 0.1:
        assessment = "一般"
    else:
        assessment = "无用"
    
    print(f"{feature_name:<10} {ig_val:<12.4f} {normalized_ig:<12.4f} {assessment:<10}")

# 决策树节点选择演示
print(f"\n=== 决策树节点选择演示 ===")
best_feature = max(features, key=lambda x: information_gain(x[1], grades)[0])
print(f"最佳分割特征: '{best_feature[0]}'")
print(f"信息增益: {information_gain(best_feature[1], grades)[0]:.4f} bits")

# 环境设置和库导入
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from collections import Counter
import math
from scipy.special import entr
import warnings
warnings.filterwarnings('ignore')

# 设置中文字体（如果遇到显示问题可以注释掉）
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

def entropy(probabilities):
    """
    计算概率分布的信息熵
    
    Parameters:
    probabilities: 概率列表，应该和为1
    
    Returns:
    entropy_value: 信息熵值（比特）
    """
    # 输入验证
    probabilities = np.array(probabilities)
    if not np.allclose(np.sum(probabilities), 1.0):
        raise ValueError("概率之和必须为1")
    if np.any(probabilities < 0):
        raise ValueError("概率不能为负")
    
    entropy_val = 0.0
    for p in probabilities:
        if p > 0:  # 避免log(0)的情况
            entropy_val -= p * math.log2(p)
    return entropy_val

def cross_entropy(p, q):
    """
    计算交叉熵 H(P, Q)
    
    Parameters:
    p: 真实分布P
    q: 预测分布Q
    
    Returns:
    cross_entropy_value: 交叉熵值
    """
    p = np.array(p)
    q = np.array(q)
    
    # 输入验证
    if len(p) != len(q):
        raise ValueError("分布P和Q必须有相同的长度")
    if not np.allclose(np.sum(p), 1.0) or not np.allclose(np.sum(q), 1.0):
        raise ValueError("分布P和Q必须和为1")
    if np.any(p < 0) or np.any(q < 0):
        raise ValueError("概率不能为负")
    
    ce = 0.0
    for i in range(len(p)):
        if p[i] > 0:
            if q[i] > 0:
                ce -= p[i] * math.log2(q[i])
            else:
                return float('inf')  # 如果Q在P有概率的地方为0，交叉熵为无穷大
    return ce

class LanguageModelAnalyzer:
    """语言模型分析器，使用信息论工具"""
    
    def __init__(self, vocab_size=50000):
        self.vocab_size = vocab_size
        
    def calculate_perplexity(self, cross_entropy):
        """
        计算困惑度
        
        Parameters:
        cross_entropy: 交叉熵值
        
        Returns:
        perplexity: 困惑度
        """
        return 2 ** cross_entropy
    
    def analyze_model_uncertainty(self, predictions):
        """
        分析模型预测的不确定性
        
        Parameters:
        predictions: 模型输出的概率分布列表
        
        Returns:
        analysis: 不确定性分析结果
        """
        entropies = [entropy(pred) for pred in predictions]
        cross_entropies = [cross_entropy([1.0] + [0.0]*(len(pred)-1), pred) for pred in predictions]  # 假设完美预测
        
        analysis = {
            'average_entropy': np.mean(entropies),
            'average_cross_entropy': np.mean(cross_entropies),
            'perplexity': self.calculate_perplexity(np.mean(cross_entropies)),
            'entropy_std': np.std(entropies),
            'max_entropy': np.max(entropies),
            'min_entropy': np.min(entropies),
            'confidence_scores': [1 - e / math.log2(len(pred)) for e, pred in zip(entropies, predictions)]  # 置信度
        }
        return analysis
    
    def temperature_sampling_analysis(self, base_probs, temperatures):
        """
        分析温度采样对分布的影响
        
        Parameters:
        base_probs: 基础概率分布
        temperatures: 温度值列表
        
        Returns:
        results: 不同温度下的分析结果
        """
        results = {}
        
        for temp in temperatures:
            # 应用温度采样
            if temp == 0:
                # 贪婪采样（实际中temp不会为0）
                scaled_probs = base_probs
            else:
                scaled_probs = self.apply_temperature(base_probs, temp)
            
            results[temp] = {
                'distribution': scaled_probs,
                'entropy': entropy(scaled_probs),
                'max_prob': np.max(scaled_probs),
                'effective_tokens': self.calculate_effective_tokens(scaled_probs)
            }
        
        return results
    
    def apply_temperature(self, probs, temperature):
        """应用温度到概率分布"""
        scaled_probs = np.log(probs) / temperature
        scaled_probs = np.exp(scaled_probs - np.max(scaled_probs))  # 数值稳定性
        return scaled_probs / np.sum(scaled_probs)
    
    def calculate_effective_tokens(self, probs):
        """计算有效token数（基于熵）"""
        h = entropy(probs)
        return 2 ** h

# 大语言模型分析示例
print("=== 大语言模型信息论分析 ===")
analyzer = LanguageModelAnalyzer(vocab_size=50000)

# 模拟语言模型输出（不同确定性的预测）
sample_predictions = [
    # 高确定性预测
    [0.9, 0.05, 0.03, 0.01, 0.01] + [0.0] * 15,
    # 中等确定性预测  
    [0.4, 0.3, 0.2, 0.05, 0.05] + [0.0] * 15,
    # 低确定性预测
    [0.2, 0.15, 0.15, 0.1, 0.1, 0.08, 0.07, 0.05, 0.05, 0.05] + [0.0] * 10,
    # 均匀分布（高不确定性）
    [0.05] * 20
]

# 确保概率和为1
sample_predictions = [p / np.sum(p) for p in sample_predictions]

# 分析模型不确定性
uncertainty_analysis = analyzer.analyze_model_uncertainty(sample_predictions)

print("模型不确定性分析:")
print(f"  平均熵: {uncertainty_analysis['average_entropy']:.4f} bits")
print(f"  平均交叉熵: {uncertainty_analysis['average_cross_entropy']:.4f} bits")
print(f"  困惑度: {uncertainty_analysis['perplexity']:.2f}")
print(f"  熵标准差: {uncertainty_analysis['entropy_std']:.4f}")
print(f"  最大熵: {uncertainty_analysis['max_entropy']:.4f}")
print(f"  最小熵: {uncertainty_analysis['min_entropy']:.4f}")
print(f"  平均置信度: {np.mean(uncertainty_analysis['confidence_scores']):.3f}")

# 温度采样分析
print(f"\n=== 温度采样分析 ===")
base_distribution = np.array([0.6, 0.2, 0.1, 0.05, 0.03, 0.02])
temperatures = [0.1, 0.5, 1.0, 1.5, 2.0]

temp_results = analyzer.temperature_sampling_analysis(base_distribution, temperatures)

print("温度对概率分布的影响:")
print(f"{'温度':<8} {'熵':<8} {'最大概率':<12} {'有效token数':<15}")
print("-" * 50)

for temp in temperatures:
    result = temp_results[temp]
    print(f"{temp:<8} {result['entropy']:<8.4f} {result['max_prob']:<12.4f} {result['effective_tokens']:<15.2f}")

# 可视化温度采样的影响
plt.figure(figsize=(15, 5))

for i, temp in enumerate(temperatures[:3]):  # 只显示前3个温度
    plt.subplot(1, 3, i+1)
    result = temp_results[temp]
    
    plt.bar(range(len(result['distribution'])), result['distribution'])
    plt.title(f'温度 = {temp}\n熵 = {result["entropy"]:.3f}')
    plt.xlabel('Token索引')
    plt.ylabel('概率')
    plt.ylim(0, 1)

plt.tight_layout()
plt.show()