📖 第11章：大模型时代的NLP¶

emergent_abilities = {
    "In-Context Learning": {
        "描述": "给几个示例，大模型就能执行新任务",
        "涌现规模": "~100B参数",
        "例子": "给3个情感分析示例，模型就能分析新文本",
    },
    "思维链推理(CoT)": {
        "描述": "通过'让我们一步步思考'引导逐步推理",
        "涌现规模": "~100B参数",
        "例子": "数学推理、逻辑推理",
    },
    "指令遵循": {
        "描述": "理解并遵循复杂的自然语言指令",
        "涌现规模": "经过指令微调后",
        "例子": "根据详细的格式要求生成内容",
    },
}

scaling_law = """
缩放定律 (Scaling Law):
  模型性能 ∝ f(模型大小, 数据量, 计算量)

  模型变大 → 性能持续提升，且在某些阈值处出现质变（涌现）

  关键参数规模:
  - ~1B: 基础生成能力
  - ~10B: 较好的理解和推理
  - ~100B: In-Context Learning、CoT涌现
  - ~1T: 多模态、复杂推理
"""
print(scaling_law)

prompt_techniques = {
    "角色设定": {
        "模板": "你是一个{角色}，你的任务是{任务描述}",
        "例子": "你是一个资深NLP工程师，请帮我分析这段文本的情感",
    },
    "任务描述": {
        "模板": "请{动作}以下{对象}，要求{约束条件}",
        "例子": "请将以下中文翻译成英文，要求准确、流畅、自然",
    },
    "输出格式指定": {
        "模板": "请按以下JSON格式返回结果：{格式模板}",
        "例子": '请返回JSON格式：{"sentiment": "正面/负面", "confidence": 0.0-1.0}',
    },
    "少样本示例": {
        "模板": "参考以下示例：\n输入：{x1}\n输出：{y1}\n\n输入：{x_new}\n输出：",
        "例子": "输入：好看\n输出：正面\n输入：难看\n输出：负面\n输入：一般\n输出：",
    },
}

# Prompt模板类
class PromptTemplate:
    """Prompt模板管理器"""

    def __init__(self):
        self.templates = {}

    def add_template(self, name, template):
        self.templates[name] = template

    def format(self, name, **kwargs):  # *args接收任意位置参数，**kwargs接收任意关键字参数
        if name not in self.templates:
            raise KeyError(f"模板'{name}'不存在")
        return self.templates[name].format(**kwargs)

# 使用
pt = PromptTemplate()

pt.add_template("sentiment", """你是一个情感分析专家。
请分析以下文本的情感倾向。

文本：{text}

请从以下选项中选择：正面、负面、中性
并给出你的分析理由。

输出格式:
情感: [正面/负面/中性]
理由: [你的分析]
""")

pt.add_template("ner", """请从以下文本中提取命名实体。

文本: {text}

请提取：
- 人名(PER)
- 地名(LOC)
- 机构名(ORG)
- 时间(TIME)

输出格式(JSON):
[{{"entity": "实体文本", "type": "实体类型"}}]
""")

pt.add_template("translate", """请将以下{src_lang}文本翻译成{tgt_lang}。
要求翻译准确、自然、流畅。

原文: {text}

翻译:""")

# 测试
print(pt.format("sentiment", text="这部电影太好看了"))
print("\n" + "="*50 + "\n")
print(pt.format("ner", text="2024年3月15日，马斯克在旧金山宣布特斯拉新品"))

advanced_prompts = {
    "自一致性(Self-Consistency)": {
        "方法": "多次采样取多数投票",
        "代码": """
# 多次采样，取出现最多的答案
answers = [llm(prompt, temperature=0.7) for _ in range(5)]
final_answer = max(set(answers), key=answers.count)
""",
    },
    "思维树(Tree-of-Thought)": {
        "方法": "探索多条推理路径，评估后选最优",
        "代码": """
# 1. 生成多个推理方案
# 2. 评估每个方案的合理性
# 3. 选择最优路径继续推理
""",
    },
    "检索增强Prompt": {
        "方法": "先检索相关知识，再构建Prompt",
        "代码": """
# 1. 检索相关文档
docs = retriever.search(question)
# 2. 构建包含检索结果的Prompt
prompt = f"参考资料:{docs}\\n问题:{question}\\n答案:"
""",
    },
}

for name, info in advanced_prompts.items():
    print(f"📌 {name}: {info['方法']}")

"""
指令微调流程：

1. 收集指令数据集
   - 人工编写的<指令,输入,输出>三元组
   - 现有数据集转化为指令格式
   - 使用更强大的模型生成（Self-Instruct）

2. 微调基础模型
   - 在指令数据集上继续训练
   - 通常用较小的学习率

3. 效果
   - 模型变得更"听话"
   - 零样本能力提升
"""

instruction_dataset_example = [
    {
        "instruction": "判断以下文本的情感倾向",
        "input": "这部电影太好看了，强烈推荐",
        "output": "正面。文本中'太好看了'和'强烈推荐'都是明显的正面情感词。",
    },
    {
        "instruction": "将以下英文翻译成中文",
        "input": "Artificial intelligence is transforming the world.",
        "output": "人工智能正在改变世界。",
    },
    {
        "instruction": "提取文本中的命名实体",
        "input": "马云在杭州创立了阿里巴巴集团",
        "output": "人名：马云\n地名：杭州\n机构名：阿里巴巴集团",
    },
    {
        "instruction": "用一句话总结以下段落的主要内容",
        "input": "随着深度学习技术的快速发展，自然语言处理领域取得了巨大突破。"
                 "从BERT到GPT-4，预训练语言模型不断刷新各项任务的性能记录。",
        "output": "深度学习推动了NLP领域的重大突破，预训练模型持续提升任务性能。",
    },
]

# 格式化指令微调数据
def format_instruction(example):
    """将指令数据格式化为训练文本"""
    if example["input"]:
        prompt = f"""### 指令:
{example['instruction']}

### 输入:
{example['input']}

### 输出:
{example['output']}"""
    else:
        prompt = f"""### 指令:
{example['instruction']}

### 输出:
{example['output']}"""
    return prompt

print("指令微调数据示例:")
for i, ex in enumerate(instruction_dataset_example[:2]):  # enumerate同时获取索引和元素  # 切片操作，取前n个元素
    print(f"\n--- 样本 {i+1} ---")
    print(format_instruction(ex))

"""
LoRA (Low-Rank Adaptation):
- 不修改原始预训练权重
- 在注意力层添加低秩矩阵 ΔW = A × B
- 原始权重W固定，只训练A和B

W_new = W + ΔW = W + A × B
  W: (d × d)  原始权重（冻结）
  A: (d × r)  低秩矩阵A（训练）
  B: (r × d)  低秩矩阵B（训练）
  r << d      秩远小于维度（如r=8）

参数效率：
  原始微调: d × d = 768 × 768 = 589,824
  LoRA: d × r + r × d = 768 × 8 + 8 × 768 = 12,288
  仅需原始的 2%！
"""

import torch
import torch.nn as nn

class LoRALinear(nn.Module):  # 继承nn.Module定义网络层
    """LoRA低秩适配层"""

    def __init__(self, original_linear, rank=8, alpha=16):
        super().__init__()  # super()调用父类方法
        self.original = original_linear
        self.rank = rank
        self.alpha = alpha

        in_features = original_linear.in_features
        out_features = original_linear.out_features

        # 低秩矩阵
        self.lora_A = nn.Parameter(torch.randn(in_features, rank) * 0.01)
        self.lora_B = nn.Parameter(torch.zeros(rank, out_features))

        # 冻结原始权重
        for param in self.original.parameters():
            param.requires_grad = False

        self.scaling = alpha / rank

    def forward(self, x):
        # 原始输出 + LoRA输出
        original_out = self.original(x)
        lora_out = (x @ self.lora_A @ self.lora_B) * self.scaling
        return original_out + lora_out

# 演示
original_linear = nn.Linear(768, 768)
lora_linear = LoRALinear(original_linear, rank=8, alpha=16)

# 参数量对比
original_params = sum(p.numel() for p in original_linear.parameters())
lora_trainable = sum(p.numel() for p in lora_linear.parameters() if p.requires_grad)

print(f"原始Linear参数量: {original_params:,}")
print(f"LoRA可训练参数量: {lora_trainable:,}")
print(f"参数比例: {lora_trainable/original_params:.2%}")

class RewardModel(nn.Module):
    """简化版奖励模型"""

    def __init__(self, base_model_dim=768):
        super().__init__()
        self.reward_head = nn.Sequential(
            nn.Linear(base_model_dim, 256),
            nn.ReLU(),
            nn.Linear(256, 1),
        )

    def forward(self, hidden_state):
        """
        hidden_state: BERT/GPT的[CLS]或最后一个token的输出
        返回标量奖励分数
        """
        return self.reward_head(hidden_state).squeeze(-1)  # squeeze压缩维度

# 奖励模型训练数据格式
preference_data = [
    {
        "prompt": "请解释什么是机器学习",
        "chosen": "机器学习是人工智能的一个分支，它让计算机系统能够从数据中学习和改进...",
        "rejected": "机器学习就是让电脑学东西。",
    },
    {
        "prompt": "1+1等于几",
        "chosen": "1+1=2。这是基本的算术加法运算。",
        "rejected": "这个问题太简单了，自己想。",
    },
]

"""
奖励模型的训练损失:
  L = -log(σ(r(chosen) - r(rejected)))

目标：让好回答的奖励分数 > 差回答的奖励分数
"""
print("RLHF奖励模型训练损失:")
print("  L = -log(σ(r_chosen - r_rejected))")

"""
DPO (Direct Preference Optimization):
不需要单独训练奖励模型和RL优化，直接用偏好数据优化策略

核心思想：
将奖励模型隐式地整合到语言模型的训练目标中

DPO损失:
L = -log σ(β(log π(y_w|x)/π_ref(y_w|x) - log π(y_l|x)/π_ref(y_l|x)))

其中:
  y_w: 偏好的回答(winner)
  y_l: 不偏好的回答(loser)
  π: 当前策略
  π_ref: 参考策略(SFT模型)
  β: 温度参数

优势：
- 不需要奖励模型
- 不需要RL训练（PPO的实现复杂且不稳定）
- 训练更稳定
- 实现更简单
"""
print("DPO vs RLHF:")
print("  RLHF: SFT → 训练RM → PPO训练（三阶段，复杂）")
print("  DPO:  SFT → DPO训练（两阶段，简单）")

# 标准Prompt vs CoT Prompt

standard_prompt = """
问题: 小明有5个苹果，给了小红3个，又买了4个，请问小明现在有几个苹果？
答案:"""

cot_prompt = """
问题: 小明有5个苹果，给了小红3个，又买了4个，请问小明现在有几个苹果？

让我们一步步思考：
1. 小明最初有5个苹果
2. 给了小红3个，剩下 5 - 3 = 2 个
3. 又买了4个，现在有 2 + 4 = 6 个

答案: 6个苹果
"""

# Zero-shot CoT
zero_shot_cot = """
问题: 小明有5个苹果，给了小红3个，又买了4个，请问小明现在有几个苹果？

让我们一步步思考：
"""

print("CoT Prompt技巧:")
print("  1. Few-shot CoT: 给出包含推理步骤的示例")
print("  2. Zero-shot CoT: 在问题后加'让我们一步步思考'")
print("  3. 复杂问题CoT显著提升性能（简单问题差异不大）")

cot_variants = {
    "标准CoT": {
        "描述": "给出推理步骤",
        "提示": "让我们一步步思考",
    },
    "Self-Consistency": {
        "描述": "多次CoT采样，取多数投票",
        "代码": "对同一问题生成5次推理链，选最常见答案",
    },
    "Tree-of-Thought (ToT)": {
        "描述": "树状搜索多条推理路径",
        "代码": "生成多个推理分支 → 评估每个分支 → 选最优路径",
        "实现示例": """
# Tree-of-Thought 实现示例
class TreeOfThought:
    def __init__(self, model, max_depth=3, num_thoughts=3):
        self.model = model
        self.max_depth = max_depth
        self.num_thoughts = num_thoughts

    def generate_thoughts(self, state, n=3):
        \"\"\"生成n个可能的下一步思考\"\"\"
        prompt = f"当前状态: {state}\\n生成{n}个可能的下一步思考:"
        return self.model.generate(prompt, num_return=n)

    def evaluate_thought(self, thought):
        \"\"\"评估思考路径的质量(1-10分)\"\"\"
        prompt = f"评估这个思考的合理性(1-10分): {thought}"
        return float(self.model.generate(prompt))

    def search(self, problem):
        \"\"\"使用BFS/DFS搜索最优推理路径\"\"\"
        # 初始化：从问题生成初始思考
        root_thoughts = self.generate_thoughts(problem, self.num_thoughts)

        best_path = []
        best_score = 0

        # BFS搜索每条路径
        for thought in root_thoughts:
            score = self.evaluate_thought(thought)
            if score > best_score:
                best_score = score
                best_path = [thought]

        return best_path

# 使用示例
# tot = TreeOfThought(llm)
# solution = tot.search("如何用24升和13升的容器量出5升水？")
""",
    },
    "Auto-CoT": {
        "描述": "自动生成思维链示例",
        "代码": "让模型自己生成推理步骤作为示例",
    },
    "Program-of-Thought (PoT)": {
        "描述": "生成Python代码而非自然语言推理",
        "代码": "将推理转化为可执行的代码",
    },
}

for name, info in cot_variants.items():
    print(f"📌 {name}: {info['描述']}")

"""
In-Context Learning (ICL):
不需要梯度更新，仅通过在prompt中提供示例就能执行新任务

Zero-shot: 没有示例
One-shot:  1个示例
Few-shot:  多个示例(通常2-5个)
"""

# Zero-shot
zero_shot = """判断文本情感：
文本：这部电影太好看了
情感："""

# One-shot
one_shot = """判断文本情感：

文本：画面精美，值得一看
情感：正面

文本：这部电影太好看了
情感："""

# Few-shot
few_shot = """判断文本情感：

文本：画面精美，值得一看
情感：正面

文本：太难看了，浪费时间
情感：负面

文本：还行，中规中矩
情感：中性

文本：这部电影太好看了
情感："""

print("ICL示例格式:")
print("  Zero-shot: 直接给任务")
print("  One-shot: 1个示例 + 新输入")
print("  Few-shot: 多个示例 + 新输入")
print("\n最佳实践:")
print("  - 示例选择很重要，选与目标问题相似的")
print("  - 示例顺序可能影响结果")
print("  - 标签分布要均衡")
print("  - 通常3-5个示例效果就很好")

task_evolution = {
    "文本分类": {
        "传统": "BERT微调 → 分类头",
        "大模型": "Prompt: '请判断这段文本的类别: ...'",
        "何时用传统": "数据多、延迟要求低、成本敏感",
    },
    "NER": {
        "传统": "BiLSTM-CRF / BERT-NER",
        "大模型": "Prompt: '请提取文本中的实体: ...'",
        "何时用传统": "对精度要求极高、生产环境部署",
    },
    "机器翻译": {
        "传统": "专用翻译模型(NLLB/MarianMT)",
        "大模型": "直接用GPT翻译",
        "何时用传统": "低资源语言、批量翻译、低成本",
    },
    "信息抽取": {
        "传统": "Pipeline/联合抽取模型",
        "大模型": "Prompt: '请从文本中提取(实体,关系,实体)三元组'",
        "何时用传统": "结构化输出要求严格、大规模处理",
    },
    "文本摘要": {
        "传统": "TextRank / BART / T5",
        "大模型": "Prompt: '请用一句话总结: ...'",
        "何时用传统": "基本被大模型取代",
    },
}

print("大模型时代NLP任务演变:")
for task, info in task_evolution.items():
    print(f"\n📌 {task}:")
    print(f"   传统方法: {info['传统']}")
    print(f"   大模型方法: {info['大模型']}")
    print(f"   传统优势场景: {info['何时用传统']}")

choice_guide = """
选择指南：

何时用传统方法（BERT微调等）：
✅ 有大量标注数据
✅ 对延迟要求严格（<10ms）
✅ 对成本敏感（大模型调用费用高）
✅ 对精度要求极高（如医疗、金融）
✅ 需要本地部署（数据隐私）
✅ 输出需要严格结构化

何时用大模型：
✅ 标注数据很少或没有
✅ 需要快速原型验证
✅ 任务需要复杂推理
✅ 多语言/跨领域场景
✅ 开放域生成任务
✅ 对延迟不太敏感
"""
print(choice_guide)

class LLMBasedNLP:
    """基于大模型API的NLP工具箱"""

    def __init__(self, model="gpt-4o-mini"):
        self.model = model

    def _call_llm(self, prompt):
        """调用LLM API（伪代码）"""
        # 实际使用时替换为真实API调用
        # import openai
        # response = openai.chat.completions.create(...)
        return f"[LLM Response for: {prompt[:50]}...]"

    def sentiment_analysis(self, text):
        """情感分析"""
        prompt = f"""请分析以下文本的情感倾向，返回JSON格式。

文本: {text}

输出格式:
{{"sentiment": "正面/负面/中性", "confidence": 0.0-1.0, "reason": "分析原因"}}"""
        return prompt

    def ner(self, text):
        """命名实体识别"""
        prompt = f"""请从以下文本中提取所有命名实体。

文本: {text}

请提取以下类型的实体:
- PER: 人名
- ORG: 组织/公司
- LOC: 地点
- TIME: 时间

输出JSON格式:
[{{"entity": "实体文本", "type": "实体类型", "start": 起始位置}}]"""
        return prompt

    def summarize(self, text, max_words=50):
        """文本摘要"""
        prompt = f"""请用不超过{max_words}个字总结以下文本的核心内容。

文本: {text}

摘要:"""
        return prompt

    def translate(self, text, src_lang="中文", tgt_lang="英文"):
        """翻译"""
        prompt = f"""请将以下{src_lang}文本翻译成{tgt_lang}。
要求翻译准确、自然、流畅。

原文: {text}

翻译:"""
        return prompt

# 使用
nlp = LLMBasedNLP()
print("大模型NLP工具箱 - Prompt示例:")
print("\n" + "="*50)
print("情感分析:")
print(nlp.sentiment_analysis("这家餐厅的菜品非常好吃"))
print("\n" + "="*50)
print("NER:")
print(nlp.ner("2024年，马斯克在旧金山发布了特斯拉的新产品"))

import json

class StructuredOutput:
    """从大模型获取结构化输出"""

    @staticmethod  # @staticmethod不需要实例即可调用
    def parse_json_response(response_text):
        """从LLM响应中提取JSON"""
        try:  # try/except捕获异常
            # 尝试直接解析
            return json.loads(response_text)  # json.loads将JSON字符串解析为Python对象
        except json.JSONDecodeError:
            # 尝试从代码块中提取
            import re
            json_match = re.search(r'```json?\s*([\s\S]*?)\s*```', response_text)  # re.search在字符串中搜索匹配模式
            if json_match:
                return json.loads(json_match.group(1))

            # 尝试找到JSON子串
            json_match = re.search(r'[\[{][\s\S]*?[\]}]', response_text)
            if json_match:
                return json.loads(json_match.group())

        return None

    @staticmethod
    def validate_schema(data, schema):
        """简单的schema验证"""
        if isinstance(schema, dict):
            if not isinstance(data, dict):
                return False
            for key in schema:
                if key not in data:
                    return False
        elif isinstance(schema, list):
            if not isinstance(data, list):
                return False
        return True

# 测试
parser = StructuredOutput()

test_responses = [
    '{"sentiment": "正面", "confidence": 0.95}',
    '```json\n{"entities": [{"text": "马斯克", "type": "PER"}]}\n```',
    '答案是 [{"name": "BERT"}]',
]

for response in test_responses:
    result = parser.parse_json_response(response)
    print(f"解析: '{response[:40]}...' → {result}")