AI智能体的全面解析:从基础到前沿,技术拆解全流程(宝藏收藏学习)
摘要:从烤串摊到AI智能体——技术拆解与实现 本文通过烤串摊场景生动拆解AI智能体技术架构,包含以下核心模块: 输入感知:使用Whisper实现语音识别,结合LLM完成结构化订单提取(NER技术) 知识检索:基于RAG架构,利用ChromaDB向量数据库存储烤串知识库和顾客历史记录 决策推理:通过LangChain构建智能体链,整合LLM推理、知识检索和记忆模块 执行反向输出包含取货位置、调料标准
⚠️ 长文警告!没耐心的请直接关闭! ⚠️
这篇文章会从烤串摊的“二狗子”讲起,但别以为是在讲段子——接下来的内容会塞满语音识别代码、RAG实现、Transformer架构、LangChain智能体搭建、QLoRA微调、FastAPI部署等硬核技术细节,全文超过一万字,代码块能让你 scroll 到手酸。
如果你连看完一个GitHub README的耐心都没有,或者觉得“调参”“写Prompt”是“低级劳动”,那这篇文章不适合你。毕竟,烤串都得一串一串穿,代码得一行一行写,AI模型得一次一次调参才能收敛——没耐心的人,做什么都不会成功的,现在关掉还来得及。
留下来的人,我们开始:从烤串摊到AI智能体,从“二狗子”到“二狗子Pro Max”,把技术拆解得明明白白。
第一章:AI智能体?不就是我的烤串摊“二狗子”吗?
先给AI智能体一个技术定义:AI智能体(Agent)是一个能感知环境(Perception)、基于内部状态和知识进行推理(Reasoning)、做出决策(Decision-Making)并执行行动(Action),同时能从反馈中学习(Learning)的计算系统。
对应到烤串摊的“二狗子”,我们把这个流程拆成四个技术模块,每个模块都上代码:
1.1 耳朵听单(输入感知:从语音到结构化数据)
顾客喊“老板,10个羊肉串,微辣,多孜然!”——这是模拟信号(语音),二狗子得先把它转成数字信号(文本),再提取成结构化订单。
技术栈:
- 自动语音识别(ASR):OpenAI Whisper(开源,多语言支持)
- 命名实体识别(NER)/ 结构化提取:LLM(GPT-4o-mini 或 LLaMA 3-8B)
代码实现:
首先,用Whisper做语音识别:
# 安装依赖:pip install openai-whisper torch
import whisper
# 加载模型(tiny/base/small/large,越大越准,越慢)
model = whisper.load_model("small")
# 模拟顾客语音输入(实际场景中是麦克风流或音频文件)
audio_file = "customer_order.wav"
result = model.transcribe(audio_file, language="zh")
print("转录文本:", result["text"])
# 输出:转录文本:老板,10个羊肉串,微辣,多孜然!
接下来,用LLM把文本转成结构化JSON(方便后续处理):
# 安装依赖:pip install langchain-openai
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
from typing import List, Optional
# 定义订单的Pydantic模型(结构化数据格式)
class OrderItem(BaseModel):
item_name: str = Field(description="烤串名称,如羊肉串、烤板筋")
quantity: int = Field(description="数量")
spiciness: str = Field(description="辣度:不辣、微辣、中辣、特辣")
extra: Optional[List[str]] = Field(description="额外要求,如多孜然、不要葱")
class CustomerOrder(BaseModel):
items: List[OrderItem]
customer_id: Optional[str] = Field(description="老顾客ID(如果识别到)")
# 初始化LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 定义输出解析器
parser = PydanticOutputParser(pydantic_object=CustomerOrder)
# 定义Prompt模板
prompt = PromptTemplate(
template="你是烤串摊的订单录入员。将顾客的文本订单转换为结构化JSON。\n"
"如果顾客是老顾客(比如提到‘上次不要葱’),请生成customer_id(比如‘customer_001’)。\n"
"{format_instructions}\n"
"顾客订单:{order_text}",
input_variables=["order_text"],
partial_variables={"format_instructions": parser.get_format_instructions()}
)
# 构建链
chain = prompt | llm | parser
# 测试
order_text = result["text"] + "(对了,我是上次不要葱的王哥)"
structured_order = chain.invoke({"order_text": order_text})
print("结构化订单:")
print(structured_order.model_dump_json(indent=2))
输出的结构化订单长这样:
{
"items": [
{
"item_name": "羊肉串",
"quantity": 10,
"spiciness": "微辣",
"extra": ["多孜然", "不要葱"]
}
],
"customer_id": "customer_001"
}
这一步的核心是多模态输入感知——除了语音,还可以是文字(微信订单)、图片(顾客拍的上次订单),技术上对应ASR、OCR(光学字符识别,用PaddleOCR或GPT-4o的视觉能力)。
1.2 脑子“脑补”(理解与决策:LLM + RAG + 记忆)
二狗子拿到结构化订单后,得做三件事:
- 理解意图:确认是“新订单”还是“加单”还是“退单”。
- 检索知识:查“羊肉串在冰柜第几层”“微辣是多少辣椒粉”(这是RAG,检索增强生成)。
- 结合记忆:查“customer_001(王哥)上次的订单记录”(这是记忆模块,向量数据库)。
技术栈:
- 大语言模型(LLM):作为推理核心
- 向量数据库(Vector DB):ChromaDB / Milvus(存储知识和历史记忆,做相似度检索)
- RAG框架:LangChain
代码实现:
首先,构建“烤串知识库”和“顾客历史订单库”,存入向量数据库:
# 安装依赖:pip install chromadb langchain-chroma langchain-openai sentence-transformers
import chromadb
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document
# 初始化Embedding模型(把文本转成向量,用于相似度检索)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# 1. 构建烤串知识库(比如食材位置、调料标准)
knowledge_docs = [
Document(page_content="羊肉串存放位置:冰柜第二层左侧", metadata={"source": "inventory_guide"}),
Document(page_content="微辣标准:辣椒粉抖2下(每下约0.5g)", metadata={"source": "spice_guide"}),
Document(page_content="多孜然标准:孜然罐摇3圈(每圈约1g)", metadata={"source": "spice_guide"}),
]
# 2. 构建顾客历史订单库
history_docs = [
Document(page_content="customer_001(王哥):2026-02-20订单,10羊肉串,微辣,不要葱,备注:对辣敏感,微辣=1下辣椒粉", metadata={"source": "order_history"}),
]
# 初始化ChromaDB向量数据库
client = chromadb.Client()
knowledge_db = Chroma.from_documents(knowledge_docs, embeddings, client=client, collection_name="knowledge")
history_db = Chroma.from_documents(history_docs, embeddings, client=client, collection_name="order_history")
接下来,用LangChain构建“理解与决策链”:
from langchain_core.runnables import RunnablePassthrough, RunnableParallel
from langchain_core.output_parsers import StrOutputParser
# 定义检索器(从向量库中检索相关知识)
knowledge_retriever = knowledge_db.as_retriever(k=2)
history_retriever = history_db.as_retriever(k=1)
# 定义Prompt模板:结合输入、知识、历史记忆,生成执行计划
decision_prompt = PromptTemplate(
template="你是烤串摊的二狗子(AI智能体)。根据以下信息生成执行计划。\n"
"当前订单:{order}\n"
"知识库参考:{knowledge}\n"
"顾客历史记录:{history}\n"
"执行计划格式:\n"
"1. 取货:[食材名称,位置]\n"
"2. 调料:[辣度(具体几下),额外要求]\n"
"3. 备注:[注意事项]",
input_variables=["order", "knowledge", "history"]
)
# 构建链:并行检索知识和历史,然后传给LLM生成决策
decision_chain = (
RunnableParallel(
order=RunnablePassthrough(),
knowledge=knowledge_retriever,
history=history_retriever
)
| decision_prompt
| llm
| StrOutputParser()
)
# 测试:传入结构化订单
execution_plan = decision_chain.invoke(structured_order.model_dump_json())
print("执行计划:")
print(execution_plan)
输出的执行计划:
1. 取货:[羊肉串,冰柜第二层左侧]
2. 调料:[辣度(1下辣椒粉,因为王哥对辣敏感),多孜然(摇3圈),不要葱]
3. 备注:王哥是老顾客,千万不要加葱
这里的核心公式就是你之前看到的:
决策=f(输入,记忆,知识库) \text{决策} = f(\text{输入}, \text{记忆}, \text{知识库}) 决策=f(输入,记忆,知识库)
技术上,fff 就是LLM的前向传播(Forward Pass),输入是拼接了订单、记忆、知识的Prompt,输出是执行计划。
1.3 手脚麻利(执行行动:工具调用与API交互)
二狗子有了执行计划,得动手干活——但AI智能体没有真的“手”,它靠工具调用(Tool Calling) 来执行:
- 查库存:调用库存管理API
- 算价格:调用计算器工具
- 下单给烤炉:调用烤炉控制API(如果是智能烤炉)
- 回复顾客:调用微信公众号API
技术栈:
- LangChain Tools:封装工具
- OpenAI Function Calling / LangChain ToolCalling:让LLM自动选择和调用工具
代码实现:
首先,定义工具(比如“查库存”“算价格”):
from langchain_core.tools import tool
# 模拟库存数据库
inventory = {
"羊肉串": 100,
"烤板筋": 50,
"烤鸡翅": 30
}
# 模拟价格表
price_list = {
"羊肉串": 3,
"烤板筋": 4,
"烤鸡翅": 8
}
# 定义工具1:查库存
@tool
def check_inventory(item_name: str) -> str:
"""查询食材库存,输入食材名称,返回库存数量"""
if item_name in inventory:
return f"{item_name}库存:{inventory[item_name]}串"
else:
return f"未找到食材:{item_name}"
# 定义工具2:算价格
@tool
def calculate_price(item_name: str, quantity: int) -> str:
"""计算订单价格,输入食材名称和数量,返回总价"""
if item_name in price_list:
total = price_list[item_name] * quantity
return f"{item_name} x {quantity} = {total}元"
else:
return f"未找到价格:{item_name}"
# 把工具打包
tools = [check_inventory, calculate_price]
# 让LLM绑定工具(自动选择调用哪个工具)
llm_with_tools = llm.bind_tools(tools)
然后,构建“执行链”,让LLM自动调用工具:
from langchain.agents import AgentExecutor, create_tool_calling_agent
# 定义Agent的Prompt
agent_prompt = PromptTemplate(
template="你是烤串摊的二狗子。根据执行计划调用工具完成任务。\n"
"执行计划:{execution_plan}\n"
"工具列表:{tools}\n"
"请先查库存,再算价格,最后生成总结。\n"
"{agent_scratchpad}", # 这里是Agent的“思考过程”,记录工具调用历史
input_variables=["execution_plan", "tools", "agent_scratchpad"]
)
# 创建Agent
agent = create_tool_calling_agent(llm_with_tools, tools, agent_prompt)
# 创建Agent执行器(负责运行Agent,处理工具调用)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
# 测试:传入执行计划
result = agent_executor.invoke({"execution_plan": execution_plan})
print("执行结果:")
print(result["output"])
运行时会看到Agent的思考过程(verbose=True):
> Entering new AgentExecutor chain...
[Agent思考] 我需要先查羊肉串的库存,再算价格。
[调用工具] check_inventory with args {"item_name": "羊肉串"}
[工具返回] 羊肉串库存:100串
[Agent思考] 库存充足,现在算价格。
[调用工具] calculate_price with args {"item_name": "羊肉串", "quantity": 10}
[工具返回] 羊肉串 x 10 = 30元
[Agent思考] 任务完成,生成总结。
> Finished chain.
执行结果:
库存充足(羊肉串100串),订单总价30元。已准备好取货和调料,注意不要加葱!
这一步的核心是工具编排——LLM是“大脑”,工具是“手脚”,Agent负责把大脑的决策转换成手脚的动作。
1.4 观察反馈(学习与优化:在线学习与模型评估)
顾客吃了烤串皱眉:“这微辣还是辣了点!”——二狗子得把这个反馈记下来,更新自己的“决策模型”。
技术栈:
- 在线学习(Online Learning):用新数据实时更新模型
- 强化学习(RL):如果反馈是“好评”(奖励+1),“差评”(奖励-1),用PPO算法更新策略
- 模型评估:用Hugging Face Evaluate库评估决策准确率
代码实现:
首先,定义“反馈学习”的简单逻辑(用梯度下降更新“辣度权重”):
import numpy as np
# 初始化“辣度模型”:王哥的微辣权重(初始是0.5,即标准微辣的一半)
spiciness_weight = {
"customer_001": 0.5
}
# 学习率α(记性好不好:太大容易忘事,太小学得慢)
alpha = 0.1
def update_spiciness_weight(customer_id, feedback):
"""
根据反馈更新辣度权重
feedback: 1=太辣(需要减辣), 0=刚好, -1=太淡(需要加辣)
"""
if customer_id in spiciness_weight:
# 核心公式:新知识 = 旧知识 + α × (反馈 - 预期)
# 这里预期是“刚好(0)”,反馈是“太辣(1)”,所以要减权重
spiciness_weight[customer_id] -= alpha * feedback
# 限制权重在0.1到1.0之间
spiciness_weight[customer_id] = np.clip(spiciness_weight[customer_id], 0.1, 1.0)
print(f"更新后 {customer_id} 的辣度权重:{spiciness_weight[customer_id]}")
# 测试:王哥反馈“太辣(feedback=1)”
update_spiciness_weight("customer_001", feedback=1)
# 输出:更新后 customer_001 的辣度权重:0.4
如果是更复杂的场景(比如让二狗子自动学烤串火候),我们可以用强化学习(PPO),简单的代码框架如下(用Stable Baselines3):
# 安装依赖:pip install stable-baselines3 gymnasium
import gymnasium as gym
from stable_baselines3 import PPO
from stable_baselines3.common.envs import DummyVecEnv
# 定义“烤串环境”(模拟烤炉,状态是温度、时间,动作是加火、减火、保持)
class KebabEnv(gym.Env):
def __init__(self):
super().__init__()
self.action_space = gym.spaces.Discrete(3) # 0=减火, 1=保持, 2=加火
self.observation_space = gym.spaces.Box(low=0, high=300, shape=(2,), dtype=np.float32) # [温度, 时间]
self.temp = 200 # 初始温度
self.time = 0 # 初始时间
def step(self, action):
# 执行动作,更新状态
if action == 0:
self.temp -= 10
elif action == 2:
self.temp += 10
self.time += 1
# 计算奖励:温度在220-240之间,时间在5-8分钟,奖励+1;否则-1
reward = 1 if (220 <= self.temp <= 240 and 5 <= self.time <= 8) else -1
done = self.time >= 10 # 10分钟后结束
return np.array([self.temp, self.time], dtype=np.float32), reward, done, False, {}
def reset(self, seed=None, options=None):
self.temp = 200
self.time = 0
return np.array([self.temp, self.time], dtype=np.float32), {}
# 初始化环境和PPO模型
env = DummyVecEnv([lambda: KebabEnv()])
model = PPO("MlpPolicy", env, verbose=1)
# 训练模型(让二狗子学烤串火候)
model.learn(total_timesteps=10000)
# 测试模型
obs = env.reset()
for _ in range(10):
action, _states = model.predict(obs)
obs, rewards, dones, info = env.step(action)
print(f"温度:{obs[0][0]}, 时间:{obs[0][1]}, 奖励:{rewards[0]}")
这一步的核心公式:
新知识=旧知识+α×(反馈−预期) \text{新知识} = \text{旧知识} + \alpha \times (\text{反馈} - \text{预期}) 新知识=旧知识+α×(反馈−预期)
技术上,这就是梯度下降(Gradient Descent) 的简化版——反馈是“损失函数(Loss)”,α是“学习率(Learning Rate)”,我们通过调整模型参数(比如辣度权重、神经网络权重)来最小化损失。
第二章:想亲手调教“二狗子”?成为“AI大厨”的进修指南!
现在你知道了“二狗子”的工作流,接下来我们从“新手村”到“大师之路”,把技术栈拆解得明明白白——别担心,每一步都有代码,跟着敲就行。
2.1 新手村:打好基础,串好人生第一串
新手村的核心是**“工具链入门”**——就像烤串得先学会穿签、生火,AI得先学会Python、数学、基础机器学习。
2.1.1 Python编程:你的“穿肉签”
Python是AI的“通用胶水”——重点学这三块:
1. 基础语法 + 异步编程(Asyncio)
AI应用很多是异步的(比如同时处理10个顾客订单),所以得学asyncio:
import asyncio
import time
# 同步处理(一个一个来,慢)
def sync_process_order(order_id):
print(f"处理订单 {order_id}...")
time.sleep(1) # 模拟处理时间
print(f"订单 {order_id} 完成")
# 异步处理(同时来,快)
async def async_process_order(order_id):
print(f"处理订单 {order_id}...")
await asyncio.sleep(1) # 异步等待
print(f"订单 {order_id} 完成")
# 测试同步
start = time.time()
for i in range(3):
sync_process_order(i)
print(f"同步耗时:{time.time() - start:.2f}秒") # 输出约3.0秒
# 测试异步
start = time.time()
async def main():
tasks = [async_process_order(i) for i in range(3)]
await asyncio.gather(*tasks)
asyncio.run(main())
print(f"异步耗时:{time.time() - start:.2f}秒") # 输出约1.0秒
2. 数据处理:NumPy + Pandas
管理烤串库存、分析销量,全靠这俩:
import numpy as np
import pandas as pd
# NumPy:矩阵运算(比如计算“辣度矩阵”)
spiciness_matrix = np.array([
[0.5, 0.6, 0.7], # 王哥的微辣、中辣、特辣权重
[1.0, 1.0, 1.0] # 新顾客的权重
])
print("辣度矩阵:\n", spiciness_matrix)
# Pandas:表格处理(比如管理库存表)
inventory_df = pd.DataFrame({
"item_name": ["羊肉串", "烤板筋", "烤鸡翅"],
"stock": [100, 50, 30],
"price": [3, 4, 8]
})
print("库存表:\n", inventory_df)
# 保存和读取CSV
inventory_df.to_csv("inventory.csv", index=False)
df = pd.read_csv("inventory.csv")
3. API调用:Requests + httpx
让二狗子去“隔壁摊借酱油”——调用第三方API:
# 安装依赖:pip install requests httpx
import requests
import httpx
# 同步调用(Requests)
def sync_call_api():
response = requests.get("https://api.example.com/weather?city=Beijing")
return response.json()
# 异步调用(httpx,更快)
async def async_call_api():
async with httpx.AsyncClient() as client:
response = await client.get("https://api.example.com/weather?city=Beijing")
return response.json()
2.1.2 数学小灶:别怕,就“亿”点点
数学是AI的“底层逻辑”——不用怕,我们只学“有用的”,而且用代码实现。
1. 线性代数:向量与矩阵
AI中的数据(比如顾客订单、图像像素)都是用向量/矩阵表示的:
import numpy as np
# 向量:一个顾客订单的特征 [羊肉串数量, 辣度, 等待时间]
order_vector = np.array([10, 0.5, 5]) # 0.5是微辣权重
# 矩阵:多个顾客的订单
order_matrix = np.array([
[10, 0.5, 5], # 王哥
[5, 1.0, 3], # 李姐
[8, 0.8, 7] # 张叔
])
# 矩阵运算:计算订单相似度(点积)
similarity = np.dot(order_matrix[0], order_matrix[1])
print("王哥和李姐的订单相似度:", similarity)
2. 概率统计:理解“脑补”的不确定性
顾客说“随便”,到底多大概率要辣?用贝叶斯推断:
# 安装依赖:pip install pymc3
import pymc3 as pm
import numpy as np
# 模拟数据:100个说“随便”的顾客,60个要了辣
data = np.array([1]*60 + [0]*40) # 1=要辣,0=不要辣
# 贝叶斯模型:推断P(辣|随便)
with pm.Model() as model:
# 先验:假设概率在0-1之间均匀分布
p = pm.Uniform("p", lower=0, upper=1)
# 似然:用伯努利分布拟合数据
obs = pm.Bernoulli("obs", p=p, observed=data)
# 采样:用MCMC找后验分布
trace = pm.sample(1000, tune=1000)
# 查看结果
pm.summary(trace)
# 输出:p的均值约0.6,95%置信区间[0.5, 0.7]
3. 微积分:梯度下降的手动实现
“学习”的本质是“找最小值”——用梯度下降找“烤串损失函数”的最小值:
import numpy as np
import matplotlib.pyplot as plt
# 定义损失函数:比如“烤串焦糊度”随温度变化的函数
def loss_function(temp):
return (temp - 230)**2 # 最佳温度是230度,离得越远损失越大
# 定义梯度(损失函数的导数)
def gradient(temp):
return 2 * (temp - 230)
# 梯度下降初始化
temp = 200 # 初始温度
learning_rate = 0.1 # 学习率
epochs = 50 # 迭代次数
# 记录过程
temp_history = [temp]
loss_history = [loss_function(temp)]
# 梯度下降循环
for _ in range(epochs):
temp -= learning_rate * gradient(temp) # 核心:更新参数
temp_history.append(temp)
loss_history.append(loss_function(temp))
# 可视化
plt.plot(loss_history)
plt.xlabel("迭代次数")
plt.ylabel("损失")
plt.title("梯度下降过程")
plt.show()
print("最终温度:", temp) # 输出约230度
2.1.3 机器学习初体验:让机器学会“看火候”
用Scikit-learn做基础模型——比如“预测明天卖多少串”:
# 安装依赖:pip install scikit-learn matplotlib
from sklearn.linear_model import LinearRegression
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_squared_error
import numpy as np
import pandas as pd
# 模拟数据:日期、温度、周末、销量
data = pd.DataFrame({
"day": range(1, 31),
"temp": np.random.randint(15, 30, 30),
"is_weekend": np.random.randint(0, 2, 30),
"sales": 50 + 2*data["temp"] + 30*data["is_weekend"] + np.random.randn(30)*5
})
# 特征和标签
X = data[["temp", "is_weekend"]]
y = data["sales"]
# 划分训练集和测试集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)
# 训练模型
model = LinearRegression()
model.fit(X_train, y_train)
# 预测
y_pred = model.predict(X_test)
# 评估
mse = mean_squared_error(y_test, y_pred)
print("均方误差:", mse)
print("模型系数:", model.coef_) # 温度每升1度,销量加2;周末销量加30
2.2 进阶区:掌握核心科技,定制秘制酱料
进阶区的核心是**“深度学习 + LLM + 智能体框架”**——就像烤串得学会调秘制酱料,AI得学会搭神经网络、用LLM、做智能体。
2.2.1 深度学习:解锁“脑补”神功
用PyTorch搭神经网络——比如“烤串火候分类模型”(判断“生”“熟”“焦”):
1. PyTorch基础:张量与自动微分
import torch
# 张量:PyTorch的“向量/矩阵”
x = torch.tensor([2.0, 3.0], requires_grad=True) # requires_grad=True表示要计算梯度
y = x[0] * x[1] + x[0]**2 # 定义计算图
# 自动微分:计算y对x的梯度
y.backward()
print("梯度:", x.grad) # 输出:tensor([7., 2.])(因为dy/dx0 = x1 + 2x0 = 3+4=7,dy/dx1 = x0 = 2)
2. 搭建Transformer编码器(LLM的核心组件)
Transformer是LLM的“心脏”——我们手动搭一个简化版:
import torch
import torch.nn as nn
import torch.nn.functional as F
class MultiHeadAttention(nn.Module):
"""多头自注意力机制(Transformer的核心)"""
def __init__(self, d_model, num_heads):
super().__init__()
self.d_model = d_model
self.num_heads = num_heads
self.d_k = d_model // num_heads
self.Wq = nn.Linear(d_model, d_model)
self.Wk = nn.Linear(d_model, d_model)
self.Wv = nn.Linear(d_model, d_model)
self.Wo = nn.Linear(d_model, d_model)
def forward(self, x):
batch_size = x.size(0)
# 线性变换 + 分头
q = self.Wq(x).view(batch_size, -1, self.num_heads, self.d_k).transpose(1, 2)
k = self.Wk(x).view(batch_size, -1, self.num_heads, self.d_k).transpose(1, 2)
v = self.Wv(x).view(batch_size, -1, self.num_heads, self.d_k).transpose(1, 2)
# 计算注意力分数
scores = torch.matmul(q, k.transpose(-2, -1)) / torch.sqrt(torch.tensor(self.d_k, dtype=torch.float32))
attn = F.softmax(scores, dim=-1)
# 聚合值
out = torch.matmul(attn, v).transpose(1, 2).contiguous().view(batch_size, -1, self.d_model)
return self.Wo(out)
class TransformerEncoderLayer(nn.Module):
"""Transformer编码器层"""
def __init__(self, d_model, num_heads, d_ff, dropout=0.1):
super().__init__()
self.self_attn = MultiHeadAttention(d_model, num_heads)
self.ffn = nn.Sequential(
nn.Linear(d_model, d_ff),
nn.ReLU(),
nn.Linear(d_ff, d_model)
)
self.norm1 = nn.LayerNorm(d_model)
self.norm2 = nn.LayerNorm(d_model)
self.dropout = nn.Dropout(dropout)
def forward(self, x):
# 自注意力 + 残差连接 + 层归一化
attn_out = self.self_attn(x)
x = self.norm1(x + self.dropout(attn_out))
# 前馈网络 + 残差连接 + 层归一化
ffn_out = self.ffn(x)
x = self.norm2(x + self.dropout(ffn_out))
return x
# 测试:输入一个batch的序列(batch_size=2, seq_len=5, d_model=128)
x = torch.randn(2, 5, 128)
encoder_layer = TransformerEncoderLayer(d_model=128, num_heads=8, d_ff=512)
out = encoder_layer(x)
print("输出形状:", out.shape) # 输出:torch.Size([2, 5, 128])
2.2.2 LLM与AI智能体开发:打造“二狗子Pro Max”
现在我们用LangChain和AutoGen做真正的AI智能体——比如“多智能体烤串店”(接单员、烤串师、收银员分工协作)。
1. LangChain智能体:带记忆和RAG的“二狗子Pro”
我们把第一章的内容整合起来,做一个完整的LangChain智能体:
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_core.tools import tool
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain.prompts import PromptTemplate
from langchain.memory import ConversationBufferMemory
# 1. 初始化组件
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
# 2. 构建向量库(知识库+历史订单)
knowledge_docs = [
Document(page_content="羊肉串存放位置:冰柜第二层左侧", metadata={"source": "inventory_guide"}),
Document(page_content="微辣标准:辣椒粉抖2下", metadata={"source": "spice_guide"}),
]
history_docs = [
Document(page_content="customer_001(王哥):不要葱,微辣=1下辣椒粉", metadata={"source": "order_history"}),
]
knowledge_db = Chroma.from_documents(knowledge_docs, embeddings, collection_name="knowledge")
history_db = Chroma.from_documents(history_docs, embeddings, collection_name="order_history")
knowledge_retriever = knowledge_db.as_retriever(k=2)
history_retriever = history_db.as_retriever(k=1)
# 3. 定义工具
@tool
def check_inventory(item_name: str) -> str:
"""查询库存"""
return f"{item_name}库存:100串"
@tool
def calculate_price(item_name: str, quantity: int) -> str:
"""计算价格"""
return f"{item_name} x {quantity} = {quantity*3}元"
tools = [check_inventory, calculate_price]
# 4. 定义Prompt(结合记忆、RAG、工具)
prompt = PromptTemplate(
template="你是烤串摊的二狗子Pro。\n"
"聊天历史:{chat_history}\n"
"知识库参考:{knowledge}\n"
"顾客历史:{history}\n"
"工具列表:{tools}\n"
"用户输入:{input}\n"
"{agent_scratchpad}",
input_variables=["chat_history", "knowledge", "history", "input", "tools", "agent_scratchpad"]
)
# 5. 构建Agent
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, memory=memory, verbose=True)
# 6. 测试
def get_knowledge(input):
return knowledge_retriever.invoke(input)
def get_history(input):
return history_retriever.invoke(input)
# 这里用RunnableParallel把knowledge和history传进去
from langchain_core.runnables import RunnableParallel, RunnablePassthrough
chain = (
RunnableParallel(
input=RunnablePassthrough(),
knowledge=lambda x: get_knowledge(x["input"]),
history=lambda x: get_history(x["input"])
)
| agent_executor
)
result = chain.invoke({"input": "我是王哥,来10个羊肉串,微辣"})
print(result["output"])
2. AutoGen多智能体:“接单员-烤串师-收银员”协作
AutoGen让多个智能体分工协作——就像烤串店有不同的员工:
# 安装依赖:pip install pyautogen
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
# 1. 配置LLM
llm_config = {
"config_list": [{"model": "gpt-4o-mini", "api_key": "your-api-key"}]
}
# 2. 定义智能体
# 接单员:负责接订单、确认需求
order_taker = AssistantAgent(
name="接单员",
system_message="你是烤串店的接单员。负责接收顾客订单,确认需求,然后转给烤串师。",
llm_config=llm_config
)
# 烤串师:负责烤串、控制火候
kebab_master = AssistantAgent(
name="烤串师",
system_message="你是烤串店的烤串师。根据订单烤串,完成后通知收银员。",
llm_config=llm_config
)
# 收银员:负责算价格、收款
cashier = AssistantAgent(
name="收银员",
system_message="你是烤串店的收银员。负责计算订单价格,通知顾客付款。",
llm_config=llm_config
)
# 用户代理:模拟顾客
user_proxy = UserProxyAgent(
name="顾客",
human_input_mode="NEVER", # 不需要人工输入,自动模拟
max_consecutive_auto_reply=10,
is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("完成")
)
# 3. 定义群聊(让智能体在群里协作)
groupchat = GroupChat(
agents=[user_proxy, order_taker, kebab_master, cashier],
messages=[],
max_round=10
)
manager = GroupChatManager(groupchat=groupchat, llm_config=llm_config)
# 4. 启动对话
user_proxy.initiate_chat(manager, message="我是王哥,来10个羊肉串,微辣,不要葱")
运行后会看到智能体之间的对话:
顾客(to 管理器):我是王哥,来10个羊肉串,微辣,不要葱
接单员(to 群聊):收到王哥的订单:10个羊肉串,微辣,不要葱。请烤串师准备。
烤串师(to 群聊):收到订单,开始烤串:10个羊肉串,微辣(1下辣椒粉),不要葱。
烤串师(to 群聊):烤串完成!请收银员算价格。
收银员(to 群聊):订单价格:10x3=30元。王哥请付款。
收银员(to 群聊):完成
3. 提示工程(Prompt Engineering):调教AI的核心秘籍
Prompt是AI的“指令书”——好的Prompt让AI变聪明,差的Prompt让AI变傻。核心公式:
优秀Prompt=清晰角色+具体任务+期望格式+示例(Few-Shot) \text{优秀Prompt} = \text{清晰角色} + \text{具体任务} + \text{期望格式} + \text{示例(Few-Shot)} 优秀Prompt=清晰角色+具体任务+期望格式+示例(Few-Shot)
代码示例:Few-Shot + Chain-of-Thought(思维链)Prompt
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 定义Prompt:角色+任务+示例+格式
prompt = PromptTemplate(
template="你是烤串店的二狗子,一个专业的AI智能体。\n"
"任务:根据顾客订单生成执行计划。\n"
"示例:\n"
"顾客:我是李姐,来5个烤板筋,中辣\n"
"执行计划:\n"
"1. 思考:李姐是新顾客,中辣=标准中辣(3下辣椒粉)\n"
"2. 取货:烤板筋,冰柜第三层右侧\n"
"3. 调料:中辣(3下辣椒粉),无额外要求\n"
"4. 备注:无\n"
"现在处理以下订单:\n"
"顾客:{customer_order}\n"
"执行计划:",
input_variables=["customer_order"]
)
chain = prompt | llm
# 测试
result = chain.invoke({"customer_order": "我是王哥,来10个羊肉串,微辣,不要葱"})
print(result.content)
输出的执行计划会包含“思考过程”(Chain-of-Thought),让AI的决策更可解释。
2.3 大师之路:秘制配方,开宗立派
大师之路的核心是**“定制化 + 部署 + 安全”**——就像烤串得有自己的秘制配方、开分店,AI得微调专属模型、部署上线、保证安全。
2.3.1 模型微调(Finetuning):定制专属“二狗子”
通用LLM不懂你的烤串店——用QLoRA(量化低秩适应) 微调一个专属模型,低成本、高效率。
代码示例:用QLoRA微调LLaMA 3-8B
# 安装依赖:pip install transformers peft accelerate datasets bitsandbytes
import torch
from datasets import load_dataset
from transformers import (
AutoModelForCausalLM,
AutoTokenizer,
BitsAndBytesConfig,
TrainingArguments,
)
from peft import LoraConfig, PeftModel
from trl import SFTTrainer
# 1. 配置量化(4位量化,节省显存)
bnb_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_use_double_quant=True,
bnb_4bit_quant_type="nf4",
bnb_4bit_compute_dtype=torch.bfloat16
)
# 2. 加载模型和Tokenizer
model_name = "meta-llama/Meta-Llama-3-8B"
model = AutoModelForCausalLM.from_pretrained(
model_name,
quantization_config=bnb_config,
device_map="auto",
trust_remote_code=True,
use_auth_token=True
)
tokenizer = AutoTokenizer.from_pretrained(model_name, use_auth_token=True)
tokenizer.pad_token = tokenizer.eos_token
# 3. 配置LoRA
lora_config = LoraConfig(
r=16, # LoRA秩
lora_alpha=32, # 缩放因子
target_modules=["q_proj", "v_proj"], # 要微调的模块
lora_dropout=0.05,
bias="none",
task_type="CAUSAL_LM"
)
# 4. 加载数据集(你的烤串店订单数据)
# 数据集格式:{"text": "顾客:我是王哥,来10个羊肉串... 执行计划:..."}
dataset = load_dataset("json", data_files="kebab_dataset.json")
# 5. 配置训练参数
training_args = TrainingArguments(
output_dir="./kebab-llama3",
per_device_train_batch_size=4,
gradient_accumulation_steps=4,
learning_rate=2e-5,
num_train_epochs=3,
logging_steps=10,
save_strategy="epoch",
fp16=True,
push_to_hub=False
)
# 6. 初始化SFTTrainer(监督微调)
trainer = SFTTrainer(
model=model,
train_dataset=dataset["train"],
args=training_args,
tokenizer=tokenizer,
peft_config=lora_config,
dataset_text_field="text",
max_seq_length=512
)
# 7. 开始微调
trainer.train()
# 8. 保存模型
trainer.model.save_pretrained("./kebab-llama3-final")
tokenizer.save_pretrained("./kebab-llama3-final")
微调后,模型会更懂你的烤串店——比如知道“王哥的微辣是1下辣椒粉”。
2.3.2 部署上线:把“二狗子”放到网上
用FastAPI 把微调好的模型部署成API,让顾客可以通过微信、网页下单:
代码示例:FastAPI部署
# 安装依赖:pip install fastapi uvicorn transformers peft
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from transformers import AutoModelForCausalLM, AutoTokenizer
from peft import PeftModel
# 1. 初始化FastAPI
app = FastAPI(title="二狗子烤串店API", version="1.0")
# 2. 加载微调好的模型
model_path = "./kebab-llama3-final"
model = AutoModelForCausalLM.from_pretrained(model_path, device_map="auto")
tokenizer = AutoTokenizer.from_pretrained(model_path)
# 3. 定义请求和响应格式
class OrderRequest(BaseModel):
customer_order: str
class OrderResponse(BaseModel):
execution_plan: str
# 4. 定义API端点
@app.post("/generate_plan", response_model=OrderResponse)
async def generate_plan(request: OrderRequest):
try:
# 构建Prompt
prompt = f"顾客:{request.customer_order}\n执行计划:"
inputs = tokenizer(prompt, return_tensors="pt").to("cuda")
# 生成执行计划
outputs = model.generate(**inputs, max_new_tokens=200, temperature=0)
execution_plan = tokenizer.decode(outputs[0], skip_special_tokens=True).split("执行计划:")[-1]
return OrderResponse(execution_plan=execution_plan)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# 5. 运行API
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
部署后,你可以用curl测试:
curl -X POST "http://localhost:8000/generate_plan" -H "Content-Type: application/json" -d '{"customer_order": "我是王哥,来10个羊肉串,微辣"}'
2.3.3 AI安全与伦理:做负责任的“AI大厨”
最后,也是最重要的——AI安全。你得保证“二狗子”不胡说八道(幻觉)、不泄露顾客隐私、不做有害的事。
1. 幻觉检测:用RAG减少幻觉
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 定义Prompt:让AI只根据知识库回答,不知道就说“不知道”
prompt = PromptTemplate(
template="你是烤串店的二狗子。只根据以下知识库回答问题,不知道就说‘对不起,我不知道’。\n"
"知识库:{knowledge}\n"
"问题:{question}",
input_variables=["knowledge", "question"]
)
chain = prompt | llm
# 测试
knowledge = "羊肉串存放位置:冰柜第二层左侧"
question = "烤板筋在哪里?"
result = chain.invoke({"knowledge": knowledge, "question": question})
print(result.content) # 输出:对不起,我不知道
2. 隐私保护:用差分隐私(Differential Privacy)
# 安装依赖:pip install opacus
import torch
from opacus import PrivacyEngine
# 模拟模型和数据
model = torch.nn.Linear(10, 2)
optimizer = torch.optim.SGD(model.parameters(), lr=0.01)
data_loader = torch.utils.data.DataLoader(
torch.utils.data.TensorDataset(torch.randn(100, 10), torch.randint(0, 2, (100,))),
batch_size=10
)
# 加入差分隐私
privacy_engine = PrivacyEngine()
model, optimizer, data_loader = privacy_engine.make_private(
module=model,
optimizer=optimizer,
data_loader=data_loader,
noise_multiplier=1.0,
max_grad_norm=1.0
)
# 训练循环
for inputs, labels in data_loader:
optimizer.zero_grad()
outputs = model(inputs)
loss = torch.nn.functional.cross_entropy(outputs, labels)
loss.backward()
optimizer.step()
# 查看隐私预算
epsilon = privacy_engine.get_epsilon(delta=1e-5)
print(f"隐私预算 ε:{epsilon:.2f}")
第三章:保持幽默,享受旅程!
学习AI的过程,就像烤串摊刚开业时——烟熏火燎、手忙脚乱,不是烤糊了就是盐放多了。但记住这几点:
-
“脑补”是核心,但别信它“啥都懂”
LLM只是基于概率预测,有时会一本正经地胡说八道(幻觉)——比如告诉你“羊肉串放在冰箱顶层”(其实在第二层)。你得用RAG、人工审核来把关,就像烤串得你自己尝一口咸淡。 -
动手!动手!动手!
光看菜谱学不会烤串,光看论文学不会AI。多写代码,多跑模型,多调参——遇到报错?恭喜你,这是系统在给你“加经验值”呢!把CUDA out of memory当成“烤炉温度太高”,把Loss doesn't converge当成“盐放多了”,调整一下再来。 -
社区是你的“烧烤一条街”
GitHub、Stack Overflow、Hugging Face、知乎——藏着无数“秘方”和热心“摊主”。别闭门造车,遇到问题去搜搜,或者提个Issue,总有人愿意帮你。 -
保持好奇,保持幽默
看到AI把猫P成面包,别光笑,想想背后的原理(比如Diffusion Model的噪声添加);看到AI又出“迷惑行为”,别慌,这可能就是下一个研究热点。理性看待AI,不神话,不恐慌,它只是工具——就像你的烤炉,用得好能烤出美味的串,用不好会烤糊。
结语:
AI智能体不是什么天外飞仙,它就是人类智慧的延伸——一个超级能“脑补”、能“动手”、能“学习”的帮手。从烤串摊的“二狗子”到“二狗子Pro Max”,技术栈其实很清晰:
- 输入感知:ASR、OCR、API
- 理解与决策:LLM、RAG、向量数据库
- 执行行动:工具调用、LangChain、AutoGen
- 学习与优化:梯度下降、强化学习、模型微调
- 部署与安全:FastAPI、Docker、差分隐私
现在,拿起你的“代码签”,开始串你的“AI烤串”吧!未来,是属于那些既能深刻理解技术,又能用幽默和创造力将其融入生活的人的。祝你早日成为“AI烧烤界”的扛把子!
最后,给你一个动手清单,现在就开始:
- 跑一遍第一章的Whisper语音识别代码
- 用LangChain搭一个简单的RAG智能体
- 用FastAPI把你的智能体部署成API
就从这三步开始——记住,没耐心的人做什么都不会成功,但留下来的人,总能烤出属于自己的“美味AI烤串”。
更多推荐

所有评论(0)