简介 这两年的AI 变化真的是太大了,近半年的到现在AI已经发展的越来越厉害,可以满足很多复杂和重复的工作,他的很多知识层面已经远远超过了专业的个人技术层面的专业人员
近半年来,我在渗透测试和代码审计工作中,已经深度使用AI,基本实现了全程AI辅助
现在没有学过AI的人可能大部分还停留在和AI陪聊阶段,现在在实际工作生活,已经可以帮我我们大大减轻工作和学习效率
后面会持续更新AI相关的学习总结和笔记
本篇就先从AI 的基础概念开始
阅读目录
◦ Tool Use / Function Calling ◦ Multi-Head Attention(多头注意力) ◦ Positional Encoding(位置编码) AI 类型 根据输入输出模态和核心能力,可以将 AI 简单分为:
理解、生成图像和视频,用于 OCR、识别、绘图、视频生成 GPT、Midjourney、Stable Diffusion
除了基础模型能力之外,AI还可以进一步组成不同的应用形态:
AI 的基础内容 Prompt(提示词/提示) 提示词是什么:就是你给 AI 的输入指令,用来告诉它你是谁,要做什么、怎么做、输出什么 其实这个对AI的要操作至关重要,你说的内容和AI要操作的内容是关键
我们在开发Agent的时候至关重要,基本上可以说提示词决定一切
我们在大部分情况下只需要给AI下指令,这个下指令很关键
比如: 案例1
这是什么攻击? 案例2
你是一名网络安全分析师,请分析下面 HTTP 请求,判断是否存在 SQL 注入,并给出攻击证据、攻击类型、风险等级,最后以 JSON 格式输出 这两个完全不一样,你输入的内容 我这边写了两个代码
def simple_prompt ( http_request ): """简单提问""" prompt = f""" 这是什么攻击? {http_request} """ return call_ai(prompt) 输出如下
代码
def structured_prompt ( http_request ): """结构化指令""" prompt = f""" 你是一名网络安全分析师。 请分析下面的 HTTP 请求: 1. 判断是否存在 SQL 注入 2. 给出攻击证据 3. 判断攻击类型 4. 判断风险等级 5. 最后以 JSON 格式输出 HTTP 请求: {http_request} """ return call_ai(prompt) 简单演示完毕
Tokenizer(分词器)和 Token 一个Token不是一个字那是什么如下介绍
比如我输入一个内容是:帮我分析这个HTTP请求
Tokenizer(分词器)他会叫我们输入的内容进行拆分
帮我分析这个HTTP请求可能会拆分成
帮我 / 分析 / 这个 / HTTP / 请求 然后每一个都会变成一个独立的编号如下:
帮我 = 12345 分析 = 6789 HTTP = 1024 请求 = 5432 每一个数字就是Token的ID 我用一个
代码演示
from transformers import AutoTokenizer tokenizer = AutoTokenizer.from_pretrained( "bert-base-chinese" ) text = "hello world" tokens = tokenizer.tokenize(text) token_ids = tokenizer.encode(text, add_special_tokens= False ) print ( "Tokens:" , tokens) print ( "Token IDs:" , token_ids) print ( "Token 数量:" , len (token_ids)) 注意 :不同大模型可能使用不同的Tokenizer所以你发一个内容每个大模型的Token消耗都不同
Embedding(向量嵌入) Embedding 可以简单理解为:将信息转换成数字向量,使模型能够通过数学运算处理这些信息,并可以用于表示和比较不同信息之间的关系
对于文本来说,Token会经过Embedding转换成向量
Context Window(一次最大处理) Context Window(上下文窗口)指模型一次请求中能够处理的最大Token数量,例如,一个模型的上下文窗口为128KToken,那么一次请求能够处理的上下文长度上限大约为 128KToken,需要注意的是,不同模型的上下文窗口大小不同,而且输入Token和模型输出 Token通常都会占用上下文窗口
Context(上下文) 在 LLM 里,上下文就是模型在当前这一次请求中能够看到并参考的全部信息
注意:上下文不是单独的聊天记录
代码演示
def call_ai ( messages ): response = requests.post( BASE_URL, headers={ "Authorization" : f"Bearer {API_KEY} " , "Content-Type" : "application/json" , }, json={ "model" : MODEL, "messages" : messages, "temperature" : 0 }, timeout= 60 ) response.raise_for_status() return response.json()[ "choices" ][ 0 ][ "message" ][ "content" ] messages = [ { "role" : "system" , "content" : "用户的名字叫张三。" }, { "role" : "user" , "content" : "我叫什么名字?" } ] 看一下结果
Hallucination(幻觉) Hallucination(幻觉):AI 生成了看起来合理,语言流畅,但实际上是错误、虚构或没有依据的信息
Temperature(温度) Temperature 越低,回答越稳定、确定;越高,回答越随机、发散
RAG 检索增强 RAG :先去外部知识库找资料,再把找到的资料交给 LLM,让 LLM 根据资料回答
未添加索引时:
添加索引后:
简单方便理解代码演示一下
# 知识库 knowledge_base = [ "发现 SQL 注入漏洞后,第一步应该记录攻击请求和攻击时间。" , "确认 SQL 注入漏洞后,需要确认受到影响的接口。" , "高危安全事件需要及时通知安全负责人。" , "漏洞确认后,需要对受影响系统进行风险评估。" , ] question = "发现 SQL 注入漏洞后第一步应该做什么?" def retrieve ( question ): for document in knowledge_base: if "SQL 注入" in question and "SQL 注入" in document: return document return None # RAG context = retrieve(question) if context is None : context = "知识库中没有找到相关资料。" prompt = f""" 你是一名网络安全分析师。 请严格根据下面提供的资料回答问题。 【资料】 {context} 【问题】 {question} 如果资料中没有答案,请回答: 无法确认,缺少可靠资料。 """ # ========================= # 6. 调用 LLM # ========================= response = requests.post( BASE_URL, headers={ "Authorization" : f"Bearer {API_KEY} " , "Content-Type" : "application/json" }, json={ "model" : MODEL, "messages" : [ { "role" : "user" , "content" : prompt } ], "temperature" : 0 }, timeout= 60 ) response.raise_for_status() data = response.json() answer = data[ "choices" ][ 0 ][ "message" ][ "content" ] print ( "LLM 最终回答" ,answer) AI Agent 智能体 Tool Use / Function Calling 让LLM不只是说,而是能够请求程序帮它执行一个具体函数
LLM本身通常不会直接执行你的Python函数,而是告诉你的程序“我想调用哪个函数、传什么参数,然后你的程序真正执行
代码演示
# 告诉 AI 有什么工具 tools = [{ "type" : "function" , "function" : { "name" : "get_weather" , "description" : "查询城市天气" , "parameters" : { "type" : "object" , "properties" : { "city" : { "type" : "string" , "description" : "城市名称" } }, "required" : [ "city" ] } } }] messages = [ { "role" : "user" , "content" : "北京今天天气怎么样?" } ] # 第一次请求:让 AI 判断是否需要工具 response = requests.post( BASE_URL, headers={ "Authorization" : f"Bearer {API_KEY} " , "Content-Type" : "application/json" }, json={ "model" : MODEL, "messages" : messages, "tools" : tools, "temperature" : 0 } ) data = response.json() message = data[ "choices" ][ 0 ][ "message" ] print ( "AI 第一次回答:" ) print (message) # AI 要求调用工具 if "tool_calls" in message: tool_call = message[ "tool_calls" ][ 0 ] args = json.loads( tool_call[ "function" ][ "arguments" ] ) result = get_weather(args[ "city" ]) # 把 AI 的请求加入上下文 messages.append(message) # 把工具结果返回给 AI messages.append({ "role" : "tool" , "tool_call_id" : tool_call[ "id" ], "content" : result }) # 第二次请求:AI 根据工具结果回答 response = requests.post( BASE_URL, headers={ "Authorization" : f"Bearer {API_KEY} " , "Content-Type" : "application/json" }, json={ "model" : MODEL, "messages" : messages, "tools" : tools, "temperature" : 0 } ) answer = response.json() print ( "\nAI 最终回答:" ) print ( answer[ "choices" ][ 0 ][ "message" ][ "content" ] ) ReAct(推理 + 行动) ReAct对于Agent非常核心的一种工作模式
让 AI 不只是想答案,而是思考->调用工具->看结果 ->再思考->再行动,直到完成任务
代码演示
# 工具 def calculator ( expression ): return str ( eval (expression, { "__builtins__" : {}})) # ReAct question = "计算 123 * 456,然后告诉我结果。" messages = [ { "role" : "user" , "content" : question } ] while True : # 1. 思考 / 决策 response = requests.post( BASE_URL, headers={ "Authorization" : f"Bearer {API_KEY} " , "Content-Type" : "application/json" }, json={ "model" : MODEL, "messages" : messages, "temperature" : 0 }, timeout= 60 ) message = response.json()[ "choices" ][ 0 ][ "message" ] print ( "\nAI:" ) print (message.get( "content" , "" )) if "tool_call" not in message.get( "content" , "" ).lower(): print ( "\n最终答案:" ) print (message.get( "content" , "" )) break expression = "123 * 456" print ( "\nAction:calculator" ) print ( "Expression:" , expression) # 3. 执行工具 result = calculator(expression) print ( "Observation:" , result) # 4. 把结果返回给 AI messages.append(message) messages.append({ "role" : "user" , "content" : f"工具 calculator 返回结果: {result} ,请继续分析并给出最终答案。" }) 当前主流 Agent 框架如下:
Reflection(反思) ReAct是想->做->看结果,Reflection 是做完->检查->改进
代码演示
def ask_ai ( prompt ): response = requests.post( BASE_URL, headers={ "Authorization" : f"Bearer {API_KEY} " , "Content-Type" : "application/json" }, json={ "model" : MODEL, "messages" : [ { "role" : "user" , "content" : prompt } ], "temperature" : 0 }, timeout= 60 ) return response.json()[ "choices" ][ 0 ][ "message" ][ "content" ] # 1. 用户需求 question = "查询今天登录失败的次数" # 2. 第一次生成 sql = ask_ai( f""" 请根据需求生成 SQL,只输出 SQL。 需求: {question} """ ) print ( "第一次生成:" ) print (sql) # 3. Reflection:让 AI 检查 check = ask_ai( f""" 请检查下面的 SQL 是否正确。 需求: {question} SQL: {sql} 检查: 1. SQL 语法 2. 是否满足需求 3. 是否存在明显逻辑问题 如果有问题,请说明问题。 如果没有问题,请回答:正确 """ ) print ( "\nReflection 检查:" ) print (check) # 4. 发现问题 → 修改 if "正确" not in check: sql = ask_ai( f""" 请根据 Reflection 的检查结果修改 SQL。 原始需求: {question} 原 SQL: {sql} 检查结果: {check} 请输出修改后的 SQL,不要解释。 """ ) print ( "\n修改后的 SQL:" ) print (sql) else : print ( "\nSQL 检查通过:" ) print (sql) Plan-and-Execute(规划执行) 先让AI对复杂任务进行拆解和规划,制定清晰的执行步骤,再由执行器按照计划逐步完成任务,并最终汇总结果
代码演示
def ask ( prompt ): r = requests.post( BASE_URL, headers={ "Authorization" : f"Bearer {API_KEY} " }, json={ "model" : MODEL, "messages" : [{ "role" : "user" , "content" : prompt}], "temperature" : 0 } ) return r.json()[ "choices" ][ 0 ][ "message" ][ "content" ] question = "分析 IP 10.10.10.20 是否存在攻击行为" # Planner:AI 制定计划 plan = ask( f""" 你是 Planner。 请为下面任务制定 4 个执行步骤,只输出步骤: {question} """ ) print ( "=== Planner ===" ) print (plan) # Executor:按照计划执行 result = ask( f""" 你是 Executor。 任务: {question} 执行计划: {plan} 请按照计划进行分析,并给出最终报告。 """ ) print ( "\n=== Executor ===" ) print (result) Workflow(工作流) 提前规定好任务应该按照什么步骤执行
MCP协议 一种让AI以统一标准发现、调用和使用外部工具与数据的协议
代码演示:
当前函数是直接返回的内容,没有调用实际的工具
def ai ( prompt ): r = requests.post( BASE_URL, headers={ "Authorization" : f"Bearer {API_KEY} " }, json={ "model" : MODEL, "messages" : [{ "role" : "user" , "content" : prompt}], "temperature" : 0 } ) return r.json()[ "choices" ][ 0 ][ "message" ][ "content" ] def ip_query ( ip ): return f" {ip} IP 情报:暂无恶意记录" def log_search ( ip ): return f" {ip} 日志:发现 3 次异常访问" question = "帮我分析 172.31.29.235 是否存在攻击" # AI 决定应该使用什么工具 decision = ai( f""" 你是安全分析 AI。 用户问题: {question} 可用工具: - ip_query:查询 IP 情报 - log_search:查询访问日志 请告诉我应该调用哪个工具,只回答: ip_query 或者 log_search """ ) print ( "AI 决定:" , decision) # 根据 AI 的决定调用工具 if "ip_query" in decision: result = ip_query( "172.31.29.235" ) else : result = log_search( "172.31.29.235" ) # 工具结果再次交给 AI answer = ai( f""" 用户问题: {question} 工具返回: {result} 请根据工具结果回答用户。 """ ) print ( "最终回答:" ) print (answer) 数学时间到!!!!! Softmax 把模型输出的一堆分数,转换成总和为 1 的概率
公式
第一步:对每个分数做 e 的指数 假设:
苹果 = 2.5 香蕉 = 1.8 牛奶 = 0.3 计算:
e²·⁵ ≈ 12.18 e¹·⁸ ≈ 6.05 e⁰·³ ≈ 1.35 第二步:全部加起来 12.18 + 6.05 + 1.35 = 19.58 第三步:每个数字除以总和 苹果: 12.18 / 19.58 ≈ 0.622 香蕉: 6.05 / 19.58 ≈ 0.309 牛奶: 1.35 / 19.58 ≈ 0.069 所以:
苹果 → 62.2% 香蕉 → 30.9% 牛奶 → 6.9% 加起来:
62.2% + 30.9% + 6.9% = 100% 用python代码算,使用numpy,这个库是专门做做科学计算和数值计算的库
import numpy as np logits = np.array([ 2.5 , # 苹果 1.8 , # 香蕉 0.3 # 牛奶 ]) exp_logits = np.exp(logits) probabilities = exp_logits / np. sum (exp_logits) print (probabilities) print ( "概率总和:" , np. sum (probabilities)) 模型训练概念 这边为了很好的理解我单纯用py做演示
本节大纲:
准备训练文本 例如我们准备大量文本:
我喜欢吃苹果 我喜欢吃香蕉 我喜欢喝牛奶 小明喜欢吃苹果 小明喜欢喝牛奶 生成词表 import numpy as np texts = [ [ "我" , "喜欢" , "吃" , "苹果" ], [ "我" , "喜欢" , "吃" , "香蕉" ], [ "我" , "喜欢" , "喝" , "牛奶" ], [ "小明" , "喜欢" , "吃" , "苹果" ], [ "小明" , "喜欢" , "喝" , "牛奶" ] ] # 建立词表 words = sorted ( set (word for text in texts for word in text)) word_to_id = { word: i for i, word in enumerate (words) } id_to_word = { i: word for word, i in word_to_id.items() } vocab_size = len (words) print ( "词表:" ) print (word_to_id) print (id_to_word) print (vocab_size) 运行结果
词表: {'吃': 0, '喜欢': 1, '喝': 2, '小明': 3, '我': 4, '牛奶': 5, '苹果': 6, '香蕉': 7} {0: '吃', 1: '喜欢', 2: '喝', 3: '小明', 4: '我', 5: '牛奶', 6: '苹果', 7: '香蕉'} 8 构造训练样本 import numpy as np texts = [ [ "我" , "喜欢" , "吃" , "苹果" ], [ "我" , "喜欢" , "吃" , "香蕉" ], [ "我" , "喜欢" , "喝" , "牛奶" ], [ "小明" , "喜欢" , "吃" , "苹果" ], [ "小明" , "喜欢" , "喝" , "牛奶" ] ] word_to_id = { '吃' : 0 , '喜欢' : 1 , '喝' : 2 , '小明' : 3 , '我' : 4 , '牛奶' : 5 , '苹果' : 6 , '香蕉' : 7 } id_to_word = { 0 : '吃' , 1 : '喜欢' , 2 : '喝' , 3 : '小明' , 4 : '我' , 5 : '牛奶' , 6 : '苹果' , 7 : '香蕉' } vocab_size = 8 training_data = [] for text in texts: token_ids = [word_to_id[word] for word in text] for i in range ( 1 , len (token_ids)): x = token_ids[:i] y = token_ids[i] training_data.append((x, y)) for x, y in training_data: print ( [id_to_word[i] for i in x], id_to_word[y] ) 随机初始化 Embedding 代码
import numpy as np texts = [ [ "我" , "喜欢" , "吃" , "苹果" ], [ "我" , "喜欢" , "吃" , "香蕉" ], [ "我" , "喜欢" , "喝" , "牛奶" ], [ "小明" , "喜欢" , "吃" , "苹果" ], [ "小明" , "喜欢" , "喝" , "牛奶" ] ] word_to_id = { '吃' : 0 , '喜欢' : 1 , '喝' : 2 , '小明' : 3 , '我' : 4 , '牛奶' : 5 , '苹果' : 6 , '香蕉' : 7 } id_to_word = { 0 : '吃' , 1 : '喜欢' , 2 : '喝' , 3 : '小明' , 4 : '我' , 5 : '牛奶' , 6 : '苹果' , 7 : '香蕉' } vocab_size = 8 embedding_dim = 3 embedding = np.random.randn( vocab_size, embedding_dim ) * 0.1 print (embedding) 因为你的词表一共有8个Token,所以是8个Token乘3
建立权重矩阵 为了让模型能够预测下一个词,我们再建立一个权重矩阵
代码
import numpy as np texts = [ [ "我" , "喜欢" , "吃" , "苹果" ], [ "我" , "喜欢" , "吃" , "香蕉" ], [ "我" , "喜欢" , "喝" , "牛奶" ], [ "小明" , "喜欢" , "吃" , "苹果" ], [ "小明" , "喜欢" , "喝" , "牛奶" ] ] word_to_id = { '吃' : 0 , '喜欢' : 1 , '喝' : 2 , '小明' : 3 , '我' : 4 , '牛奶' : 5 , '苹果' : 6 , '香蕉' : 7 } id_to_word = { 0 : '吃' , 1 : '喜欢' , 2 : '喝' , 3 : '小明' , 4 : '我' , 5 : '牛奶' , 6 : '苹果' , 7 : '香蕉' } vocab_size = 8 embedding_dim = 3 W = np.random.randn( embedding_dim, vocab_size ) * 0.1 b = np.zeros(vocab_size) print ( "W" ,W) print ( "b" ,b) 看一下结果
w=如下:
b=[0. 0. 0. 0. 0. 0. 0. 0.]
一共有 8 个,因为最终要预测8个Token
前向计算 都齐全了,下面可以输入内容了
假设用户输入:我
代码计算
import numpy as np texts = [ [ "我" , "喜欢" , "吃" , "苹果" ], [ "我" , "喜欢" , "吃" , "香蕉" ], [ "我" , "喜欢" , "喝" , "牛奶" ], [ "小明" , "喜欢" , "吃" , "苹果" ], [ "小明" , "喜欢" , "喝" , "牛奶" ] ] word_to_id = { '吃' : 0 , '喜欢' : 1 , '喝' : 2 , '小明' : 3 , '我' : 4 , '牛奶' : 5 , '苹果' : 6 , '香蕉' : 7 } id_to_word = { 0 : '吃' , 1 : '喜欢' , 2 : '喝' , 3 : '小明' , 4 : '我' , 5 : '牛奶' , 6 : '苹果' , 7 : '香蕉' } vocab_size = 8 embedding_dim = 3 # 随机初始化 Embedding embedding = np.random.randn( vocab_size, embedding_dim ) * 0.1 word = "我" input_id = word_to_id[word] input_vector = embedding[input_id] print ( "Token:" , word) print ( "TokenID:" , input_id) print ( "Embedding:" , input_vector) Embedding值是: [-0.0060238 0.24530145 -0.00281051]
使用 Softmax 代码
import numpy as np texts = [ [ "我" , "喜欢" , "吃" , "苹果" ], [ "我" , "喜欢" , "吃" , "香蕉" ], [ "我" , "喜欢" , "喝" , "牛奶" ], [ "小明" , "喜欢" , "吃" , "苹果" ], [ "小明" , "喜欢" , "喝" , "牛奶" ] ] word_to_id = { '吃' : 0 , '喜欢' : 1 , '喝' : 2 , '小明' : 3 , '我' : 4 , '牛奶' : 5 , '苹果' : 6 , '香蕉' : 7 } id_to_word = { 0 : '吃' , 1 : '喜欢' , 2 : '喝' , 3 : '小明' , 4 : '我' , 5 : '牛奶' , 6 : '苹果' , 7 : '香蕉' } vocab_size = 8 embedding_dim = 3 # 随机初始化 Embedding embedding = np.random.randn( vocab_size, embedding_dim ) * 0.1 print (embedding) word = "我" input_id = word_to_id[word] input_vector = embedding[input_id] W = np.random.randn( embedding_dim, vocab_size ) * 0.1 b = np.zeros(vocab_size) logits = input_vector @ W + b def softmax ( x ): x = x - np. max (x) exp_x = np.exp(x) return exp_x / np. sum (exp_x) probabilities = softmax(logits) print ( "预测概率:" ) for i, p in enumerate (probabilities): print ( f" {id_to_word[i]} → {p: .2 %} " ) target_id = word_to_id[ "喜欢" ] loss = -np.log(probabilities[target_id]) print ( "正确答案:" , id_to_word[target_id]) print ( "正确答案概率:" , probabilities[target_id]) print ( "Loss:" , loss) 看结果:不是很理想,因为当前
训练 1000 次(梯度下降) 代码
import numpy as np texts = [ [ "我" , "喜欢" , "吃" , "苹果" ], [ "我" , "喜欢" , "吃" , "香蕉" ], [ "我" , "喜欢" , "喝" , "牛奶" ], [ "小明" , "喜欢" , "吃" , "苹果" ], [ "小明" , "喜欢" , "喝" , "牛奶" ] ] words = sorted ( set ( word for text in texts for word in text )) word_to_id = { word: i for i, word in enumerate (words) } id_to_word = { i: word for word, i in word_to_id.items() } vocab_size = len (words) print ( "词表:" ) print (word_to_id) print () training_data = [] for text in texts: token_ids = [ word_to_id[word] for word in text ] for i in range ( 1 , len (token_ids)): # 输入 x = token_ids[:i] # 正确答案 y = token_ids[i] training_data.append((x, y)) print ( "训练数据:" ) for x, y in training_data: print ( [id_to_word[i] for i in x], id_to_word[y] ) print () embedding_dim = 3 embedding = np.random.randn( vocab_size, embedding_dim ) * 0.1 W = np.random.randn( embedding_dim, vocab_size ) * 0.1 b = np.zeros(vocab_size) def softmax ( x ): x = x - np. max (x) exp_x = np.exp(x) return exp_x / np. sum (exp_x) learning_rate = 0.1 epochs = 1000 for epoch in range (epochs): total_loss = 0 for x, y in training_data: vectors = embedding[x] # 多个 Token 的 Embedding 做平均 input_vector = np.mean( vectors, axis= 0 ) logits = input_vector @ W + b probabilities = softmax(logits) loss = -np.log( probabilities[y] + 1e-10 ) d_logits = probabilities.copy() d_logits[y] -= 1 d_W = np.outer( input_vector, d_logits ) d_b = d_logits d_input = W @ d_logits W -= learning_rate * d_W b -= learning_rate * d_b d_embedding = ( d_input / len (x) ) for token_id in x: embedding[token_id] -= ( learning_rate * d_embedding ) if epoch % 100 == 0 : print ( f"Epoch {epoch} , " f"Loss: {total_loss: .4 f} " ) word = "我" input_id = word_to_id[word] # 获取“我”的 Embedding input_vector = embedding[input_id] # 重新预测 logits = input_vector @ W + b probabilities = softmax(logits) print ( "\n输入:" , word) print ( "\n预测概率:" ) for i, p in enumerate (probabilities): print ( f" {id_to_word[i]} → {p: .2 %} " ) prediction_id = np.argmax( probabilities ) prediction = id_to_word[ prediction_id ] print ( "\n模型预测的下一个词:" ) print (prediction) 看一下结果:
Transformer原理 Transformer是一种基于Attention注意力机制的神经网络架构,是现代大语言模型的重要基础架构
例如:
我 喜欢 吃 苹果 模型在理解苹果时,需要知道它和:
吃 喜欢 我 我们先了解一下三种架构
Transformer的三种经典架构 Transformer由Google在2017年提出(论文《Attention Is All You Need》),核心创新是 Self-Attention(自注意力机制) ,取代了此前Seq2Seq模型中常用的RNN/LSTM结构
◆ 1. Encoder-only 代表模型有BERT、RoBERTa、ALBERT
BERT采用Encoder-only架构,能够同时利用Token左右两侧的上下文信息,因此具有较强的文本理解能力,更适合文本分类、实体识别、语义匹配等任务
◆ 2. Decoder-only 代表模型有GPT、LLaMA、Qwen
GPT采用Decoder-only架构,通过因果自注意力(Causal Self-Attention),只能关注当前Token及其之前的内容,并通过预测下一个Token的方式逐步生成文本
◆ 3. Encoder-Decoder 代表模型有T5、BART
T5采用Encoder-Decoder架构,由Encoder负责理解输入内容,再由Decoder根据Encoder提供的信息生成输出,因此适合翻译、摘要、问答等文本转换任务
其中T5更强调将不同任务统一成Text-to-Text(文本到文本)的形式
Self-Attention(自注意力机制) ◆ 为什么要用Attention? 在传统RNN中,模型按顺序读取单词,距离越远的词越难建立联系,例如:
小明 昨天 在 公园 里 遇到 了 他 的 小学 同学 要让模型理解“同学”和“小明”的关系,RNN需要经过很多步,信息容易丢失
而Self-Attention可以让任意两个位置的词直接建立联系,不管它们隔得多远
◆ Self-Attention的计算过程 假设输入序列: 我 喜欢 吃 苹果
第一步:生成Q、K、V 对于每个Token,通过三个不同的权重矩阵生成三个向量:
输入 Embedding:X = [x₁, x₂, x₃, x₄] Q = X · W_Q # 每个 Token 生成一个 Query K = X · W_K # 每个 Token 生成一个 Key V = X · W_V # 每个 Token 生成一个 Value "
简单记忆: Q是“问题”,K是“标签”,V是“内容”
第二步:计算注意力分数 以吃为例,它会使用自己的 Query,与我、喜欢、吃、苹果的Key分别计算相关性,从而决定应该从哪些Token中获取更多信息,实际上,序列中的每个 Token 都会执行这一过程
分数 = Q(吃) · K(我)ᵀ 分数 = Q(吃) · K(喜欢)ᵀ 分数 = Q(吃) · K(吃)ᵀ 分数 = Q(吃) · K(苹果)ᵀ 第三步:Softmax归一化 把分数转成概率(总和为1):
attention_weights = softmax(分数 / √d_k) "
除以√d_k是为了防止点积随维度增大而过大,避免softmax进入梯度极小的区域
第四步:加权求和 output(吃) = attention_weights(我) · V(我) + attention_weights(喜欢) · V(喜欢) + attention_weights(吃) · V(吃) + attention_weights(苹果) · V(苹果) 这样,“吃”这个词的最终表示,就融合了整个句子的信息,尤其关注了“苹果”
◆ 代码演示:简化版Self-Attention import numpy as np def softmax ( x ): x = x - np. max (x, axis=- 1 , keepdims= True ) exp_x = np.exp(x) return exp_x / np. sum (exp_x, axis=- 1 , keepdims= True ) # 假设有 4 个 Token,每个 Embedding 维度为 3 X = np.random.randn( 4 , 3 ) # [我, 喜欢, 吃, 苹果] # 初始化 Q、K、V 权重(实际训练中会学习) d_k = 3 W_Q = np.random.randn( 3 , d_k) * 0.1 W_K = np.random.randn( 3 , d_k) * 0.1 W_V = np.random.randn( 3 , d_k) * 0.1 # 计算 Q、K、V Q = X @ W_Q # (4, 3) K = X @ W_K # (4, 3) V = X @ W_V # (4, 3) # 计算注意力分数 scores = Q @ K.T / np.sqrt(d_k) # (4, 4) print ( "注意力分数(归一化前):\n" , scores) # Softmax 得到注意力权重 attention_weights = softmax(scores) # (4, 4) print ( "\n注意力权重:\n" , attention_weights) # 加权求和得到输出 output = attention_weights @ V # (4, 3) print ( "\nSelf-Attention 输出:\n" , output) Multi-Head Attention(多头注意力) Transformer实际使用的是Multi-Head Attention,而不是单一的自注意力
◆ 为什么需要多头? 单头注意力可能只关注一种关系,但语言中有很多种关系:
多头注意力用多组Q、K、V,让模型同时关注不同类型的关系
◆ 多头注意力计算 import numpy as np def softmax ( x ): x = x - np. max (x, axis=- 1 , keepdims= True ) exp_x = np.exp(x) return exp_x / np. sum (exp_x, axis=- 1 , keepdims= True ) class MultiHeadAttention : def __init__ ( self, d_model, num_heads ): self .d_model = d_model self .num_heads = num_heads self .d_k = d_model // num_heads # 每个头有自己的 Q、K、V 权重 self .W_Q = np.random.randn(num_heads, d_model, self .d_k) * 0.1 self .W_K = np.random.randn(num_heads, d_model, self .d_k) * 0.1 self .W_V = np.random.randn(num_heads, d_model, self .d_k) * 0.1 self .W_O = np.random.randn(d_model, d_model) * 0.1 def forward ( self, X ): # X: (seq_len, d_model) seq_len, d_model = X.shape # 每个头独立计算 head_outputs = [] for h in range ( self .num_heads): Q = X @ self .W_Q[h] # (seq_len, d_k) K = X @ self .W_K[h] # (seq_len, d_k) V = X @ self .W_V[h] # (seq_len, d_k) scores = Q @ K.T / np.sqrt( self .d_k) attention_weights = softmax(scores) head_output = attention_weights @ V head_outputs.append(head_output) # 拼接所有头的输出 concat = np.concatenate(head_outputs, axis=- 1 ) # (seq_len, d_model) output = concat @ self .W_O # (seq_len, d_model) return output # 使用示例 d_model = 8 num_heads = 2 X = np.random.randn( 4 , d_model) # 4 个 Token,每个 8 维 mha = MultiHeadAttention(d_model, num_heads) output = mha.forward(X) print ( "Multi-Head Attention 输出形状:" , output.shape) Positional Encoding(位置编码) ◆ 为什么需要位置编码? Self-Attention本身并不知道Token在序列中的绝对位置:
我 喜欢 吃 苹果 苹果 吃 喜欢 我 虽然包含相同的Token但顺序不同,语义也完全不同
因此,Transformer 需要额外向 Token 表示中加入位置信息,
让模型知道:
◆ 位置编码公式 使用正弦和余弦函数:
PE(pos, 2i) = sin(pos / 10000^(2i/d_model)) PE(pos, 2i+1) = cos(pos / 10000^(2i/d_model)) ◆ Transformer完整结构(Decoder-Only) 以GPT为代表的Decoder-Only架构:
◆ 因果掩码(Causal Masking) Decoder-Only在预测时不能看到未来的词:
预测 "吃" 时:只能看到 "我 喜欢" 预测 "苹果" 时:只能看到 "我 喜欢 吃" 实现方式是在Attention分数中,把未来的位置设为 -inf :
import numpy as np def causal_mask ( seq_len ): """生成因果掩码矩阵""" mask = np.triu(np.ones((seq_len, seq_len)), k= 1 ) # 上三角为 1 mask = mask * - 1e9 # 把 1 变成 -inf return mask seq_len = 4 mask = causal_mask(seq_len) print ( "因果掩码(-inf 表示该位置被遮挡):\n" , mask) 训练过程回顾(结合Transformer) 回到你之前的小模型训练,如果用Transformer,流程是:
训练完成后,输入“我”,模型能预测出“喜欢”
三种架构对比总结 Encoder双向理解 + Decoder因果生成
完整训练示例 极简版Transformer预测下一个词
import numpy as np texts = [ [ "我" , "喜欢" , "吃" , "苹果" ], [ "我" , "喜欢" , "吃" , "香蕉" ], [ "我" , "喜欢" , "喝" , "牛奶" ], [ "小明" , "喜欢" , "吃" , "苹果" ], [ "小明" , "喜欢" , "喝" , "牛奶" ], ] words = sorted ( set (w for text in texts for w in text)) word_to_id = {w: i for i, w in enumerate (words)} id_to_word = {i: w for w, i in word_to_id.items()} vocab_size = len (words) d_model = 8 learning_rate = 0.05 epochs = 1500 np.random.seed( 42 ) W_embed = np.random.randn(vocab_size, d_model) * 0.1 W_Q = np.random.randn(d_model, d_model) * 0.1 W_K = np.random.randn(d_model, d_model) * 0.1 W_V = np.random.randn(d_model, d_model) * 0.1 W_out = np.random.randn(d_model, vocab_size) * 0.1 b_out = np.zeros(vocab_size) def softmax ( x ): x = x - np. max (x, axis=- 1 , keepdims= True ) exp_x = np.exp(x) return exp_x / np. sum (exp_x, axis=- 1 , keepdims= True ) def forward ( x_ids ): X = W_embed[x_ids] Q = X @ W_Q K = X @ W_K V = X @ W_V scores = Q @ K.T / np.sqrt(d_model) attn = softmax(scores) attn_out = attn @ V last = attn_out[- 1 ] logits = last @ W_out + b_out return softmax(logits) def train (): global W_embed, W_Q, W_K, W_V, W_out, b_out training_data = [] for text in texts: ids = [word_to_id[w] for w in text] for i in range ( 1 , len (ids)): training_data.append((ids[:i], ids[i])) for epoch in range (epochs): total_loss = 0 for x_ids, y_id in training_data: seq_len = len (x_ids) X = W_embed[x_ids] Q = X @ W_Q K = X @ W_K V = X @ W_V scores = Q @ K.T / np.sqrt(d_model) attn = softmax(scores) attn_out = attn @ V last = attn_out[- 1 ] logits = last @ W_out + b_out probs = softmax(logits) loss = -np.log(probs[y_id] + 1e-10 ) total_loss += loss d_logits = probs.copy() d_logits[y_id] -= 1 d_W_out = np.outer(last, d_logits) d_b_out = d_logits d_last = W_out @ d_logits d_attn = np.zeros((seq_len, seq_len)) d_attn[- 1 ] = V @ d_last d_V = np.outer(attn[- 1 ], d_last) d_scores = np.zeros_like(scores) d_scores[- 1 ] = attn[- 1 ] * (d_attn[- 1 ] - np.dot(d_attn[- 1 ], attn[- 1 ])) d_q_last = d_scores[- 1 ] @ K / np.sqrt(d_model) d_K = np.outer(d_scores[- 1 ], Q[- 1 ]) / np.sqrt(d_model) d_W_Q = np.outer(X[- 1 ], d_q_last) d_W_K = X.T @ d_K d_W_V = X.T @ d_V d_X = np.zeros_like(X) d_X += d_K @ W_K.T d_X += d_V @ W_V.T d_X[- 1 ] += W_Q @ d_q_last d_W_embed = np.zeros_like(W_embed) for i, tok in enumerate (x_ids): d_W_embed[tok] += d_X[i] W_embed -= learning_rate * d_W_embed W_Q -= learning_rate * d_W_Q W_K -= learning_rate * d_W_K W_V -= learning_rate * d_W_V W_out -= learning_rate * d_W_out b_out -= learning_rate * d_b_out if epoch % 150 == 0 : print ( f"Epoch {epoch} , Loss: {total_loss: .4 f} " ) def predict ( text ): ids = [word_to_id[w] for w in text.split()] probs = forward(ids) return id_to_word[ int (np.argmax(probs))], probs def show ( text ): pred, probs = predict(text) top = np.argsort(probs)[::- 1 ][: 3 ] top_str = ", " .join( f" {id_to_word[i]} {probs[i]* 100 : .1 f} %" for i in top) print ( f" {text} -> {pred} [ {top_str} ]" ) if __name__ == "__main__" : print ( "训练前:" ) show( "我" ) show( "我 喜欢" ) show( "小明 喜欢" ) print () print ( "训练中:" ) train() print () print ( "训练后:" ) show( "我" ) show( "我 喜欢" ) show( "我 喜欢 吃" ) show( "我 喜欢 喝" ) show( "小明 喜欢" ) show( "小明 喜欢 喝" )