邓立国Agent开发入门必读书《AI Agent智能体开发实践》全文试读-示例案例丰富且能正确运行_《ai agent 智能体开发实践》在线阅读-CSDN博客
AI Agent智能体开发实践【行情 报价 价格 评测】-京东
RAG（Retrieval-Augmented Generation，检索增强生成）是结合信息检索与生成式AI的技术框架，核心是通过检索外部知识（如文档、数据库）辅助大语言模型生成更准确、可靠的回答。其效果高度依赖“检索质量”和“生成质量”的协同，因此优化与评估是提升RAG系统性能的核心环节。
5.5.1 评估指标
在AI智能体开发中，评估指标是衡量系统性能的核心工具。下面从检索质量、生成质量和人工评估三个维度，结合技术原理与行业实践展开详解。
1. 检索质量（Recall@k、MRR和NDCG）
检索类任务（如RAG问答、搜索引擎）需从海量数据中筛选高价值信息，核心指标围绕相关性排序质量设计，主要包括Recall@k、MRR和NDCG。
1）Recall@k
定义：前k个结果中包含的相关文档数占总相关文档数的比例，用于衡量核心覆盖能力。
公式：Recall@k={前k个结果中的相关文档数}/{总相关文档数}。
案例：若用户需求包含5个相关文档，系统在前10个结果中返回3个，则Recall@10=3/5=0.6。
行业标准：在RAG系统中，通常要求Recall@5≥0.853。
局限性：仅关注覆盖范围，不考虑排序顺序。
2）MRR：首个相关结果的优先级
定义：首个相关结果的排名倒数的平均值。
案例：若三个查询的首个相关结果分别位于第2、3、1位，则MRR=(1/2+1/3+1/1)/3≈0.555。
适用场景：强调首结果质量的场景（如电商推荐），行业标准通常要求MRR≥0.73。
3）NDCG：排序质量的全局优化
定义：通过折扣累计增益（DCG）与理想排序增益（IDCG）的比值，量化排序质量。
技术细节：
【示例5.23】检索质量评估指标的实现。
import numpy as np
def recall_at_k(relevant, retrieved, k):
"""
计算Recall@k指标
:param relevant: 相关文档集合
:param retrieved: 检索到的文档集合
:param k: 前k个结果
:return: Recall@k值
"""
retrieved_k = retrieved[:k]
relevant_set = set(relevant)
retrieved_set = set(retrieved_k)
intersection = relevant_set & retrieved_set
return len(intersection) / len(relevant_set)
def mean_reciprocal_rank(relevant, retrieved):
"""
计算平均倒数排名(MRR)
:param relevant: 相关文档集合
:param retrieved: 检索到的文档集合
:return: MRR值
"""
reciprocal_ranks = []
relevant_set = set(relevant)
for doc in retrieved:
if doc in relevant_set:
rank = retrieved.index(doc) + 1
reciprocal_ranks.append(1.0 / rank)
break
return np.mean(reciprocal_ranks) if reciprocal_ranks else 0.0
def ndcg(relevant, retrieved, k):
"""
计算NDCG@k指标
:param relevant: 相关文档及其相关性分数列表[(doc_id, score)]
:param retrieved: 检索到的文档集合
:param k: 前k个结果
:return: NDCG@k值
"""
# 构建相关性分数字典
relevance_scores = {doc_id: score for doc_id, score in relevant}
# 计算DCG
dcg = 0.0
for i, doc_id in enumerate(retrieved[:k], 1):
rel_score = relevance_scores.get(doc_id, 0)
dcg += (2 ** rel_score – 1) / np.log2(i + 1)
# 计算IDCG
ideal_sorted = sorted(relevant, key=lambda x: x[1], reverse=True)
idcg = 0.0
for i, (doc_id, score) in enumerate(ideal_sorted[:k], 1):
idcg += (2 ** score – 1) / np.log2(i + 1)
return dcg / idcg if idcg > 0 else 0.0
# 测试用例
if __name__ == "__main__":
relevant_docs = ['doc1', 'doc3', 'doc5']
retrieved_docs = ['doc3', 'doc5', 'doc1', 'doc7', 'doc9']
relevant_with_scores = [('doc1', 3), ('doc3', 2), ('doc5', 1)]
print(f"Recall@3: {recall_at_k(relevant_docs, retrieved_docs, 3)}")
print(f"MRR: {mean_reciprocal_rank(relevant_docs, retrieved_docs)}")
print(f"NDCG@3: {ndcg(relevant_with_scores, retrieved_docs, 3)}")
运行代码，输出如下：
Recall@3: 1.0
MRR: 1.0
NDCG@3: 0.7591919243197319
2. 生成质量（BLEU、ROUGE、BERTScore）：语义与形式的双重校验
生成类任务（如机器翻译、对话生成）需兼顾内容准确性与表达流畅性，核心指标从表层匹配到深层语义逐步深化。
1）BLEU：基于n-gram的形式匹配
定义：通过计算生成文本与参考文本的n-gram重叠率来评估质量。
技术细节：
2）ROUGE-L：基于最长公共子序列（LCS）的召回率
定义：通过计算生成文本与参考文本的n-gram或最长公共子序列（ROUGE−Lrecall）的召回率评估质量，计算最长公共子序列LCS长度（记为LCS_len）。
应用场景：摘要生成，侧重覆盖核心信息，例如生成摘要与参考摘要的重叠单元数。
对话系统：评估回复是否包含必要信息，但可能忽略逻辑连贯性。
3）BERTScore：上下文语义的深度对齐
定义：基于BERT模型生成动态词向量，通过余弦相似度计算语义匹配度。
技术原理：
【示例5.24】生成质量评估指标实现。
from collections import Counter
import math
import numpy as np
from transformers import BertTokenizer, BertModel
import torch
def bleu(candidate, references, n=4):
"""
计算BLEU分数（带修正精确度和brevity penalty）
:param candidate: 候选句子 (str)
:param references: 参考句子列表 (list of str)
:param n: n-gram的最大长度
:return: BLEU分数
"""
# 分词
candidate_tokens = candidate.strip().split()
references_tokens = [ref.strip().split() for ref in references]
if not candidate_tokens:
return 0.0
# 计算修正的精确度 p_n
p_n = []
for i in range(1, n + 1):
# 候选句中的i-gram
candidate_ngrams = list(zip(*[candidate_tokens[j:] for j in range(i)]))
candidate_counts = Counter(candidate_ngrams)
if len(candidate_ngrams) == 0:
p_n.append(0.0)
continue
# 所有参考句中每个n-gram的最大出现次数
max_counts = Counter()
for ref_tokens in references_tokens:
ref_ngrams = list(zip(*[ref_tokens[j:] for j in range(i)]))
ref_counts = Counter(ref_ngrams)
for ngram, cnt in ref_counts.items():
max_counts[ngram] = max(max_counts[ngram], cnt)
# 匹配数
match = 0
for ngram, cnt in candidate_counts.items():
match += min(cnt, max_counts.get(ngram, 0))
precision = match / len(candidate_ngrams)
p_n.append(precision)
# 几何平均（若任一p_n为0，则log为-inf，最终为0）
log_p = sum(math.log(p) for p in p_n) / n if all(p > 0 for p in p_n) else -float('inf')
geo_mean = math.exp(log_p)
# Brevity Penalty
candidate_len = len(candidate_tokens)
ref_lens = [len(ref) for ref in references_tokens]
closest_ref_len = min(ref_lens, key=lambda r: (abs(r – candidate_len), r))
if candidate_len == 0:
bp = 0
elif candidate_len > closest_ref_len:
bp = 1
else:
bp = math.exp(1 – closest_ref_len / candidate_len)
bleu_score = bp * geo_mean
return bleu_score
def rouge_l(candidate, references):
"""
计算ROUGE-L分数（F1）
:param candidate: 候选句子
:param references: 参考句子列表
:return: ROUGE-L F1 分数
"""
def lcs(x, y):
m, n = len(x), len(y)
if m == 0 or n == 0:
return 0
dp = [[0] * (n + 1) for _ in range(m + 1)]
for i in range(1, m + 1):
for j in range(1, n + 1):
if x[i – 1] == y[j – 1]:
dp[i][j] = dp[i – 1][j – 1] + 1
else:
dp[i][j] = max(dp[i – 1][j], dp[i][j – 1])
return dp[m][n]
candidate_tokens = candidate.strip().split()
if not candidate_tokens:
return 0.0
best_f1 = 0.0
for ref in references:
ref_tokens = ref.strip().split()
if not ref_tokens:
continue
lcs_length = lcs(candidate_tokens, ref_tokens)
precision = lcs_length / len(candidate_tokens)
recall = lcs_length / len(ref_tokens)
f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0.0
best_f1 = max(best_f1, f1)
return best_f1
class BERTScorer:
def __init__(self, model_name='bert-base-uncased', device=None):
self.tokenizer = BertTokenizer.from_pretrained(model_name)
self.model = BertModel.from_pretrained(model_name)
self.model.eval()
self.device = device if device else ('cuda' if torch.cuda.is_available() else 'cpu')
self.model.to(self.device)
def get_bert_score(self, candidate, reference):
"""
计算BERTScore（基于余弦相似度）
:param candidate: 候选句子
:param reference: 参考句子
:return: 余弦相似度分数
"""
def get_embedding(text):
inputs = self.tokenizer(
text,
return_tensors='pt',
padding=True,
truncation=True,
max_length=512
).to(self.device)
with torch.no_grad():
outputs = self.model(**inputs)
# 使用[CLS]向量或token平均
# 这里使用所有token的平均
last_hidden = outputs.last_hidden_state # [1, seq_len, hidden]
embedding = last_hidden.mean(dim=1).cpu().numpy().squeeze() # [hidden_dim]
return embedding
if not candidate.strip() or not reference.strip():
return 0.0
try:
cand_emb = get_embedding(candidate)
ref_emb = get_embedding(reference)
# 余弦相似度
dot_product = np.dot(cand_emb, ref_emb)
norm = np.linalg.norm(cand_emb) * np.linalg.norm(ref_emb)
cosine_sim = dot_product / norm if norm != 0 else 0.0
return cosine_sim
except Exception as e:
print(f"BERTScore error: {e}")
return 0.0
# 测试用例
if __name__ == "__main__":
candidate = "the cat is sitting on the mat"
references = [
"the cat is on the mat",
"there is a cat sitting on the mat"
]
print(f"BLEU: {bleu(candidate, references):.4f}")
print(f"ROUGE-L: {rouge_l(candidate, references):.4f}")
bert_scorer = BERTScorer()
bert_scores = [bert_scorer.get_bert_score(candidate, ref) for ref in references]
avg_bert_score = sum(bert_scores) / len(bert_scores) if bert_scores else 0.0
print(f"BERTScore (avg): {avg_bert_score:.4f}")
运行代码，输出如下：
BLEU: 0.6402
ROUGE-L: 0.8333
BERTScore (avg): 0.9215
3. 人工评估：主观体验的客观量化
自动化指标难以完全捕捉用户体验，人工评估（Human Evaluation）需通过标准化流程与多维度设计降低主观性。
1）标准化流程设计
数据分层：
评测方法：
2）多维度评分体系
核心维度：
技术实践：
3）效率与成本平衡
混合评估：
工具支持：
【示例5.25】人工评估框架实现。
import json
from typing import List, Dict
class HumanEvaluation:
def __init__(self, evaluation_criteria: List[str]):
"""
初始化人工评估框架
:param evaluation_criteria: 评估标准列表
"""
self.criteria = evaluation_criteria
self.evaluation_data = []
def add_evaluation_task(self, task_id: str, candidate_output: str, reference_output: str = None):
"""
添加评估任务
:param task_id: 任务ID
:param candidate_output: 候选输出
:param reference_output: 参考输出(可选)
"""
task = {
"task_id": task_id,
"candidate_output": candidate_output,
"reference_output": reference_output,
"ratings": {}
}
self.evaluation_data.append(task)
def record_rating(self, task_id: str, evaluator_id: str, ratings: Dict[str, float]):
"""
记录评估者的评分
:param task_id: 任务ID
:param evaluator_id: 评估者ID
:param ratings: 评分字典 {criteria: score}
"""
for task in self.evaluation_data:
if task["task_id"] == task_id:
task["ratings"][evaluator_id] = ratings
break
def calculate_agreement(self) -> Dict[str, float]:
"""
计算评估者间一致性(简单实现)
:return: 各标准的评估者间一致性分数
"""
if not self.evaluation_data or len(self.evaluation_data[0]["ratings"]) < 2:
return {criterion: 0.0 for criterion in self.criteria}
agreement_scores = {criterion: [] for criterion in self.criteria}
for task in self.evaluation_data:
if len(task["ratings"]) >= 2:
evaluators = list(task["ratings"].keys())
for criterion in self.criteria:
scores = [task["ratings"][evaluator].get(criterion, 0) for evaluator in evaluators]
avg_diff = sum(abs(a – b) for a in scores for b in scores) / (len(scores) * (len(scores) – 1))
agreement_scores[criterion].append(1 – avg_diff / 5) # 假设评分范围是0-5
return {criterion: sum(scores)/len(scores) if scores else 0.0
for criterion, scores in agreement_scores.items()}
def get_average_scores(self) -> Dict[str, Dict[str, float]]:
"""
获取平均评分
:return: 各任务各标准的平均评分
"""
result = {}
for task in self.evaluation_data:
if task["ratings"]:
avg_scores = {criterion: 0.0 for criterion in self.criteria}
for evaluator in task["ratings"]:
for criterion, score in task["ratings"][evaluator].items():
avg_scores[criterion] += score
for criterion in avg_scores:
avg_scores[criterion] /= len(task["ratings"])
result[task["task_id"]] = avg_scores
return result
def save_to_json(self, filepath: str):
"""保存评估数据到JSON文件"""
with open(filepath, 'w') as f:
json.dump({
"criteria": self.criteria,
"evaluation_data": self.evaluation_data
}, f, indent=2)
@classmethod
def load_from_json(cls, filepath: str):
"""从JSON文件加载评估数据"""
with open(filepath, 'r') as f:
data = json.load(f)
instance = cls(data["criteria"])
instance.evaluation_data = data["evaluation_data"]
return instance
# 测试用例
if __name__ == "__main__":
# 初始化评估框架
criteria = ["fluency", "relevance", "coherence", "accuracy"]
evaluator = HumanEvaluation(criteria)
# 添加评估任务
evaluator.add_evaluation_task(
task_id="task1",
candidate_output="The cat is sitting on the mat.",
reference_output="A cat is sitting on the mat."
)
# 记录评估者评分
evaluator.record_rating("task1", "evaluator1", {
"fluency": 5,
"relevance": 4,
"coherence": 5,
"accuracy": 4
})
evaluator.record_rating("task1", "evaluator2", {
"fluency": 4,
"relevance": 5,
"coherence": 4,
"accuracy": 5
})
# 计算并输出结果
print("Average Scores:", evaluator.get_average_scores())
print("Inter-rater Agreement:", evaluator.calculate_agreement())
# 保存评估数据
evaluator.save_to_json("human_evaluation.json")
运行代码，输出如下：
Average Scores: {'task1': {'fluency': 4.5, 'relevance': 4.5, 'coherence': 4.5, 'accuracy': 4.5}}
Inter-rater Agreement: {'fluency': 0.8, 'relevance': 0.8, 'coherence': 0.8, 'accuracy': 0.8}
5.5.2 端到端优化方法
AI智能体真正意义上的“端到端优化”，是指从输入原始信息到输出最终决策/生成内容的全链路参数和结构可联合更新，而不再把检索、生成、决策等模块割裂调优。目前，业界落地的端到端优化方法可归纳为三大范式：联合训练（Joint Training）、强化学习优化（RL-based Optimization）和检索−生成协同优化（Retrieval-Generation Co-optimization）。下面给出技术拆解与最佳实践。
1. 联合训练
1）核心机制
多模块共享参数空间，通过统一目标函数实现跨模态/跨任务协同优化。其关键在于打破传统分阶段训练的隔离性，让不同模块在训练过程中动态交互，从而捕捉更复杂的语义关联。
2）技术实现与案例
（1）多模态联合预训练：
InternVL3采用原生多模态预训练范式，在单一阶段同时处理文本、图像、视频数据，通过混合模态输入（如交替出现的图文序列）实现跨模态语义对齐。其可变视觉位置编码（V2PE）机制通过递归计算位置索引，使视觉标记的位置增量小于文本标记，有效缓解了长序列上下文的对齐压力。
（2）检索−生成联合优化：
Atlas模型通过Contriever双编码器检索器与T5解码器的联合预训练，在少样本学习中实现突破。例如，在NaturalQuestions任务中仅用64个样本即可达到42.4%准确率，优于540B参数的PaLM模型。这种联合训练通过掩码语言建模、前缀语言建模等自监督任务，使检索器能精准定位与生成任务相关的文档 片段。
（3）跨任务参数共享：
RAVEN框架在视觉−语言模型中采用Fusion-in-Decoder架构，允许图像编码器与文本解码器共享部分Transformer层。实验显示，在MSCOCO图像字幕任务中，联合训练使CIDEr指标提升1分，在VQA任务中特定问题类型准确率提高近3%。
3）优势与挑战
（1）优势：减少中间表示的信息损失（如传统多模态模型中图像特征到文本空间的映射误差），提升模型泛化能力。例如，InternVL3在MMMU多模态基准测试中达到72.2分，成为开源模型新标杆。
（2）挑战：计算复杂度高，比如万象3.0（InternVL3）的78B参数模型需优化训练基础设施，需设计高效的混合精度训练策略和分布式并行方案。
【示例5.26】联合训练。
import torch
import torch.nn as nn
import torch.optim as optim
# 定义多任务模型
class MultiTaskModel(nn.Module):
def __init__(self):
super(MultiTaskModel, self).__init__()
self.shared_encoder = nn.Sequential(
nn.Linear(100, 64),
nn.ReLU(),
nn.Linear(64, 32)
)
self.task1_head = nn.Linear(32, 10) # 分类任务
self.task2_head = nn.Linear(32, 1) # 回归任务
def forward(self, x):
shared_features = self.shared_encoder(x)
return self.task1_head(shared_features), self.task2_head(shared_features)
# 初始化模型和优化器
model = MultiTaskModel()
optimizer = optim.Adam(model.parameters(), lr=0.001)
criterion1 = nn.CrossEntropyLoss() # 分类损失
criterion2 = nn.MSELoss() # 回归损失
# 模拟训练数据
inputs = torch.randn(16, 100)
labels1 = torch.randint(0, 10, (16,))
labels2 = torch.randn(16, 1)
# 训练循环
for epoch in range(10):
optimizer.zero_grad()
outputs1, outputs2 = model(inputs)
loss1 = criterion1(outputs1, labels1)
loss2 = criterion2(outputs2, labels2)
total_loss = loss1 + loss2 # 联合损失
total_loss.backward()
optimizer.step()
print(f"Epoch {epoch}, Loss: {total_loss.item():.4f}")
运行代码，输出如下：
Epoch 0, Loss: 2.9056
Epoch 1, Loss: 2.7970
Epoch 2, Loss: 2.6949
Epoch 3, Loss: 2.5997
Epoch 4, Loss: 2.5088
Epoch 5, Loss: 2.4212
Epoch 6, Loss: 2.3358
Epoch 7, Loss: 2.2529
Epoch 8, Loss: 2.1731
Epoch 9, Loss: 2.0966
2. 强化学习优化
1）核心机制
将智能体决策过程建模为策略网络，通过奖励函数引导模型生成符合人类偏好的输出。其核心在于构建有效的反馈闭环，使模型从试错中学习。
2）技术实现与案例
（1）人类反馈强化学习（RLHF）：
GPT-4通过RLHF微调，在模拟律师资格考试中排名前10%。其训练流程包括：
（2）对话系统策略优化：
某研究将PPO算法与细粒度奖励机制结合，在开放域对话任务中实现突破。
（3）动态环境适应：
在机器人路径规划中，强化学习优化可通过实时传感器数据（如激光雷达点云）动态调整策略网络参数。例如，某系统采用近端策略优化（PPO）结合注意力机制，使得在未知障碍物环境中路径规划成功率提升至92%。
3）关键技术点
（1）奖励工程：需平衡短期奖励（如生成流畅性）与长期奖励（如任务完成度）。GPT-4的RLHF通过引入“无害性”“相关性”等多维度奖励权重，有效减少有害输出。
（2）探索与利用平衡：在稀疏奖励场景（如复杂推理任务），可采用基于好奇心的探索机制，通过预测下一状态的不确定性来驱动模型主动探索。
【示例5.27】本示例展示了如何使用强化学习进行优化。
具体来说，我们实现了一个基于策略梯度（Policy Gradient）的简单智能体（采用REINFORCE算法的一个变种），并使用PyTorch框架。为了方便演示，该示例不依赖于特定环境，而是通过随机数据模拟智能体与环境的交互过程，因此可以直接运行。
import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
from collections import deque
import random
class PolicyNetwork(nn.Module):
def __init__(self, state_dim, action_dim):
super(PolicyNetwork, self).__init__()
self.fc = nn.Sequential(
nn.Linear(state_dim, 64),
nn.ReLU(),
nn.Linear(64, 32),
nn.ReLU(),
nn.Linear(32, action_dim),
nn.Softmax(dim=-1)
)
def forward(self, x):
return self.fc(x)
class RLAgent:
def __init__(self, state_dim, action_dim):
self.policy_net = PolicyNetwork(state_dim, action_dim)
self.optimizer = optim.Adam(self.policy_net.parameters(), lr=0.001)
self.gamma = 0.99
self.memory = deque(maxlen=10000)
self.batch_size = 32
def select_action(self, state):
state = torch.FloatTensor(state)
probs = self.policy_net(state)
action = torch.multinomial(probs, 1).item()
return action
def store_transition(self, state, action, reward, next_state, done):
self.memory.append((state, action, reward, next_state, done))
def train(self):
if len(self.memory) < self.batch_size:
return
batch = random.sample(self.memory, self.batch_size)
states, actions, rewards, next_states, dones = zip(*batch)
states = torch.FloatTensor(np.array(states))
actions = torch.LongTensor(actions)
rewards = torch.FloatTensor(rewards)
next_states = torch.FloatTensor(np.array(next_states))
dones = torch.FloatTensor(dones)
# 计算折扣回报（从后往前）
discounted_rewards = []
running_reward = 0
for reward, done in zip(reversed(rewards), reversed(dones)):
if done:
running_reward = 0
running_reward = reward + self.gamma * running_reward
discounted_rewards.insert(0, running_reward)
discounted_rewards = torch.FloatTensor(discounted_rewards)
# 标准化回报（提升稳定性）
discounted_rewards = (discounted_rewards – discounted_rewards.mean()) / (discounted_rewards.std() + 1e-7)
# 计算 log 概率和损失
log_probs = torch.log(self.policy_net(states).gather(1, actions.unsqueeze(1)). squeeze())
loss = -torch.mean(log_probs * discounted_rewards)
self.optimizer.zero_grad()
loss.backward()
self.optimizer.step()
# ============ 训练主循环 ============
if __name__ == "__main__":
agent = RLAgent(state_dim=4, action_dim=2)
episode_rewards = [] # 记录每个 episode 的总奖励
for episode in range(100):
state = np.random.randn(4)
total_reward = 0
for step in range(100):
action = agent.select_action(state)
next_state = state + np.random.randn(4) * 0.1
reward = np.random.randn() # 奖励服从标准正态分布
done = np.random.rand() < 0.05 # 5%概率终止
agent.store_transition(state, action, reward, next_state, done)
agent.train()
state = next_state
total_reward += reward
if done:
break
episode_rewards.append(total_reward)
# 打印进度
if (episode + 1) % 10 == 0:
avg_reward = np.mean(episode_rewards[-10:])
print(f"Episode {episode + 1}, Average Reward: {avg_reward:.2f}")
运行代码，输出如下：
Episode 10, Average Reward: -1.51
Episode 20, Average Reward: -0.99
Episode 30, Average Reward: 0.32
Episode 40, Average Reward: 0.18
Episode 50, Average Reward: 0.27
Episode 60, Average Reward: -1.67
Episode 70, Average Reward: -2.19
Episode 80, Average Reward: -0.11
Episode 90, Average Reward: -0.71
Episode 100, Average Reward: 1.78
3. 检索−生成协同优化
1）核心机制
通过外部知识库检索增强生成模型的事实性，同时利用生成模型的语义理解能力反哺检索器，形成闭环优化。其核心在于解决“参数记忆容量瓶颈”与“实时知识更新”问题。
2）技术实现与案例
（1）检索器−生成器联合预训练
Atlas模型通过双编码器检索器（Contriever）与Fusion-in-Decoder架构的T5解码器联合训练，在少样本问答任务中实现42.4%准确率（仅用64个样本），超越同等规模基线模型3%。其优化策略包括：
（2）检索自主性优化
Self-RAG引入反思字符（Reflection Tokens），使模型能自主判断是否需要检索。
（3）多模态检索增强
多向量检索器（Multi-Vector Retriever）支持文本、表格、图像等多模态数据检索。
3）优化策略
（1）检索结果重排序：采用BERT等模型对检索结果进行语义相关性打分，结合BM25传统检索算法，使Top-5检索结果准确率提升25%。
（2）生成结果反哺：将生成的高质量答案作为新文档存入知识库，形成“检索−生成−增量学习”闭环。例如，某客服系统通过此策略，3个月内知识库规模扩大40%，响应准确率提升9%。
【示例5.28】检索−生成协同优化示例。
#pip install torch transformers
#pip install huggingface_hub[hf_xet]
import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import BertModel, BertTokenizer
class RetrievalGenerator(nn.Module):
def __init__(self):
super(RetrievalGenerator, self).__init__()
self.retriever = BertModel.from_pretrained('bert-base-uncased')
self.generator = nn.Sequential(
nn.Linear(768*2, 512), # 合并检索和上下文信息
nn.ReLU(),
nn.Linear(512, 256),
nn.ReLU(),
nn.Linear(256, 128),
nn.ReLU(),
nn.Linear(128, 768) # 输出与BERT隐藏层相同维度
)
self.tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')
def forward(self, context_input, retrieval_input):
# 获取上下文表示
context_output = self.retriever(**context_input).last_hidden_state[:, 0, :]
# 获取检索结果表示
retrieval_output = self.retriever(**retrieval_input).last_hidden_state[:, 0, :]
# 合并信息并生成
combined = torch.cat([context_output, retrieval_output], dim=-1)
generated = self.generator(combined)
return generated
def generate(self, context_text, retrieval_text):
context_input = self.tokenizer(context_text, return_tensors='pt', padding=True, truncation=True)
retrieval_input = self.tokenizer(retrieval_text, return_tensors='pt', padding=True, truncation=True)
with torch.no_grad():
generated = self.forward(context_input, retrieval_input)
# 这里简化处理，实际应用中可能需要更复杂的解码策略
return generated
# 示例使用
model = RetrievalGenerator()
context = "The capital of France is"
retrieval = "Paris is the capital and most populous city of France"
output = model.generate(context, retrieval)
print(f"Generated representation shape: {output.shape}")
运行代码，输出如下：
Generated representation shape: torch.Size([1, 768])
5.5.3 常见问题与解决方案
在AI智能体开发中，检索噪声（Noisy Retrieval）、信息冗余和长上下文处理是影响智能体决策效率与输出质量的核心挑战。以下从问题本质、产生原因及技术解决方案三个方面进行详细解析。
1. 检索噪声
1）问题定义与危害
检索噪声是指智能体在从外部知识库、文档库或互联网中检索信息时，返回结果包含不相关、错误或低质量的内容。这些噪声会直接干扰智能体的推理过程，导致决策偏差（如错误回答、无效行动），甚至破坏用户信任。
2）产生原因
（1）查询意图理解偏差：用户输入模糊（如“苹果的最新动态”），智能体未能区分“水果苹果”或“苹果公司”，导致检索方向错误。
（2）检索算法缺陷：传统关键词匹配对同义词（如“医生”与“医师”）、多义词（如“Java”指编程语言或岛屿）不敏感，引发误匹配。
（3）知识库质量低：文档中存在错误信息（如过时的政策条款）、重复内容，或缺乏标注（如未区分“事实”与“观点”）。
3）解决方案
（1）提升查询理解精度：
（2）优化检索算法：从“字面匹配”到“语义匹配”：
（3）检索结果过滤与重排序：
（4）知识库清洗与增强：
【示例5.29】检索噪声过滤。
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import re
class RetrievalFilter:
def __init__(self, threshold=0.1):
self.threshold = threshold
# 使用空格分词（因为我们自己构造词列表）
self.vectorizer = TfidfVectorizer(
tokenizer=lambda x: x.split(), # 按空格切分
lowercase=False
)
def _extract_keywords(self, text):
"""
手动提取与医疗、AI 相关的关键词（模拟分词）
"""
# 定义关键词映射规则（可根据需求扩展）
keywords = []
chinese_only = ''.join(re.findall(r'[\\u4e00-\\u9fff]', text))
# 常见术语匹配
terms = [
'人工智能', 'AI', '深度学习', '机器学习',
'医疗', '医学', '诊断', '影像', '健康',
'应用', '系统', '模型', '算法', '分析'
]
for term in terms:
if term in text or \\
(term == 'AI' and 'AI' in text) or \\
(term == '人工智能' and ('AI' in text or '智能' in text)):
keywords.append(term)
# 如果没有匹配到，至少保留单字作为 fallback
if not keywords:
keywords = list(chinese_only)
return keywords
def filter_noisy_results(self, query, documents):
# 提取查询和文档的关键词
query_keywords = self._extract_keywords(query)
doc_keywords_list = [self._extract_keywords(doc) for doc in documents]
# 转为字符串供 TfidfVectorizer 使用
query_str = " ".join(query_keywords)
doc_strs = [" ".join(keywords) for keywords in doc_keywords_list]
# 合并向量化
all_texts = [query_str] + doc_strs
tfidf_matrix = self.vectorizer.fit_transform(all_texts)
# 计算相似度
query_vector = tfidf_matrix[0]
doc_vectors = tfidf_matrix[1:]
similarities = cosine_similarity(query_vector, doc_vectors)[0]
# 调试输出
print(f"\\n查询关键词: {query_str}")
print(f"文档关键词:")
for i, doc in enumerate(documents):
print(f" '{doc}' -> {doc_strs[i]}")
print(f"\\n文档相似度:")
for i, (doc, sim) in enumerate(zip(documents, similarities)):
print(f" '{doc}' -> {sim:.3f}")
# 过滤
filtered_docs = [
doc for doc, sim in zip(documents, similarities)
if sim >= self.threshold
]
return filtered_docs
# 测试用例
if __name__ == "__main__":
filter = RetrievalFilter(threshold=0.1) # 很低的阈值也 OK
test_query = "人工智能在医疗领域的应用"
test_docs = [
"机器学习算法概述",
"医疗AI诊断系统的最新进展",
"天气预报模型改进",
"深度学习在医学影像分析中的应用",
"电子商务推荐系统"
]
filtered = filter.filter_noisy_results(test_query, test_docs)
print("\\n过滤后的相关文档:")
if filtered:
for i, doc in enumerate(filtered, 1):
print(f"{i}. {doc}")
else:
print("无符合条件的相关文档。")
运行代码，输出如下：
查询关键词: 人工智能 医疗 应用
文档关键词:
'机器学习算法概述' -> 机器学习 算法
'医疗AI诊断系统的最新进展' -> 人工智能 AI 医疗 诊断 系统
'天气预报模型改进' -> 模型
'深度学习在医学影像分析中的应用' -> 深度学习 医学 影像 应用 分析
'电子商务推荐系统' -> 系统
文档相似度:
'机器学习算法概述' -> 0.000
'医疗AI诊断系统的最新进展' -> 0.472
'天气预报模型改进' -> 0.000
'深度学习在医学影像分析中的应用' -> 0.219
'电子商务推荐系统' -> 0.000
过滤后的相关文档:
1. 医疗AI诊断系统的最新进展
2. 深度学习在医学影像分析中的应用
关键技术点：
2. 信息冗余
1）问题定义与危害
信息冗余（Redundancy）是指智能体获取的信息中包含大量重复内容（如多文档描述同一事实）或不必要细节（如无关背景铺垫），导致处理效率下降（如增加计算成本），且关键信息被稀释（如用户需要“核心结论”却被冗长过程淹没）。
2）产生原因
（1）检索结果相似性高：同一事件的不同报道（如“某会议议程”的多个新闻稿）内容高度重叠。
（2）文档结构松散：非结构化文本（如博客、论坛帖子）常包含冗余描述（如反复强调同一观点）。
（3）多轮交互积累：长对话中，用户重复提及历史信息（如“之前说过的那个问题”），导致上下文冗余。
3）解决方案
（1）冗余检测与去重：
（2）信息压缩与提炼：
（3）结构化存储与检索：
（4）动态冗余控制：
【示例5.30】冗余信息去除。
from datasketch import MinHash, MinHashLSH
import jieba # 中文分词
class RedundancyRemover:
def __init__(self, threshold=0.7):
self.threshold = threshold
# 初始化LSH，用于快速查找相似文档
self.lsh = MinHashLSH(threshold=self.threshold, num_perm=128)
self.indexed_docs = {} # 存储 MinHash -> 文档内容（或ID），用于避免重复插入
self.doc_counter = 0 # 用于生成唯一键
def _create_minhash(self, text):
"""创建文本的MinHash签名"""
tokens = list(jieba.cut(text))
# 过滤空字符串或空白字符
tokens = [token.strip() for token in tokens if token.strip()]
mh = MinHash(num_perm=128)
for token in tokens:
mh.update(token.encode('utf-8'))
return mh
def remove_redundant(self, documents):
"""
去除冗余文档
参数:
documents: 待处理文档列表
返回:
去重后的文档列表
"""
unique_docs = []
for doc in documents:
mh = self._create_minhash(doc)
# 查询LSH中是否存在相似文档
results = self.lsh.query(mh)
if not results:
# 没有找到相似文档，认为是唯一的
key = f"doc_{self.doc_counter}"
self.doc_counter += 1
self.lsh.insert(key, mh)
unique_docs.append(doc)
else:
# 找到相似文档，跳过（不添加）
continue
return unique_docs
# 测试用例
if __name__ == "__main__":
# 确保已安装依赖
# pip install datasketch jieba
remover = RedundancyRemover(threshold=0.6)
test_docs = [
"人工智能将改变医疗行业",
"AI技术正在变革医疗领域",
"深度学习在计算机视觉中的应用",
"机器学习算法在图像识别中的使用",
"区块链技术的金融应用"
]
print("原始文档:")
for i, doc in enumerate(test_docs, 1):
print(f"{i}. {doc}")
unique_docs = remover.remove_redundant(test_docs)
print("\\n去重后的文档:")
for i, doc in enumerate(unique_docs, 1):
print(f"{i}. {doc}")
运行代码，输出如下：
原始文档:
1. 人工智能将改变医疗行业
2. AI技术正在变革医疗领域
3. 深度学习在计算机视觉中的应用
4. 机器学习算法在图像识别中的使用
5. 区块链技术的金融应用
去重后的文档:
1. 人工智能将改变医疗行业
2. AI技术正在变革医疗领域
3. 深度学习在计算机视觉中的应用
4. 机器学习算法在图像识别中的使用
5. 区块链技术的金融应用
关键技术点：
3. 长上下文处理
1）问题定义与危害
长上下文是指智能体需处理的文本长度超过模型上下文窗口限制（如10万Token的书籍），或虽在窗口内但过长导致模型注意力分散（如1万Token的报告中，首尾信息关联被忽略）。这会导致模型漏读关键信息、推理错误（如“前文提到A，后文却错误关联B”），或因计算量过大导致效率低下。
2）产生原因
（1）模型窗口限制：早期LLM（如GPT-3）仅支持4k Token，即使当前模型（如GPT-4 Turbo支持128k Token）支持超长文本（如50万Token的小说），但仍需特殊处理。
（2）注意力机制低效：Transformer的自注意力计算复杂度为O(n²)（n为文本长度），长文本会导致计算成本激增（如n=100000时，计算量是n=10000的100倍）。
（3）关键信息稀释：长文本中，核心信息（如“结论”）可能被大量无关内容（如“背景描述”）淹没，模型难以聚焦。
3）解决方案
（1）模型层面：扩展上下文窗口与优化注意力。
。（2）文本层面：分块与滑动窗口。
（3）关键信息提取与压缩。
（4）外部记忆与动态检索。
引入记忆机制，将长文本拆分为“记忆单元”（如段落向量），存储在外部数据库（如向量库）中，模型仅保留当前上下文，需回溯信息时动态检索记忆单元。例如：
【示例5.31】长上下文处理问题。
本示例演示如何为AI智能体构建“文件系统+可恢复压缩+按需加载”的长上下文处理方案。思路借鉴Manus的“文件系统即上下文”设计，并结合上下文压缩技巧，在128 K Token以上场景仍可保持低成本、高召回。
"""
#离线部署建议
#from modelscope.hub.snapshot_download import snapshot_download
# 下载模型到本地
#model_dir = snapshot_download('qwen/qwen-7b-chat')
#embed_dir = snapshot_download('AI-ModelScope/bge-small-zh-v1.5')
# 使用本地路径
#self.pipeline = pipeline(task=Tasks.text_generation, model=model_dir)
#self.embedder = pipeline(task=Tasks.text_embedding, model=embed_dir)
#pip install modelscope
#pip install sentence-transformers # 用于本地 embedding（可选）
#pip install modelscope langchain faiss-cpu # 或 faiss-gpu
try:
import fsspec
from fsspec.callbacks import TqdmCallback # 尝试正常导入
except (ImportError, AttributeError):
print("⚠️ 修复 fsspec.callbacks 兼容性问题…")
import fsspec
import tqdm
# 手动创建 callbacks 模块
if not hasattr(fsspec, 'callbacks'):
fsspec.callbacks = type('callbacks', (), {})
class TqdmCallback:
def __init__(self, tqdm_kwargs=None):
self.tqdm_kwargs = tqdm_kwargs or {}
self.pbar = None
def __enter__(self):
self.pbar = tqdm.tqdm(**self.tqdm_kwargs)
return self
def __exit__(self, *exc_info):
if self.pbar:
self.pbar.close()
def relative_update(self, inc=1):
if self.pbar:
self.pbar.update(inc)
fsspec.callbacks.TqdmCallback = TqdmCallback
# ================================
# 现在可以安全导入 datasets 和 ModelScope
# ================================
import os
import uuid
from pathlib import Path
from typing import List
from langchain.schema import Document
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.vectorstores import FAISS
from langchain.memory import ConversationBufferWindowMemory
try:
from modelscope.pipelines import pipeline
from modelscope.utils.constant import Tasks
from modelscope.outputs import OutputKeys
from modelscope.hub.snapshot_download import snapshot_download
except ImportError as e:
raise ImportError("请安装 ModelScope: pip install modelscope") from e
# ================================
# 0. 参数配置
# ================================
CHUNK_SIZE = 800 # 降低块大小以节省内存
CHUNK_OVERLAP = 100
TOP_K = 4
PERSIST_DIR = Path("./context_fs")
PERSIST_DIR.mkdir(exist_ok=True)
# 模型设置（推荐使用 1.8B 以适配低资源）
EMBEDDING_MODEL_ID = "AI-ModelScope/bge-small-zh-v1.5"
LLM_MODEL_ID = "qwen/qwen-1_8b-chat" # 更轻量
# 设备设置：使用 GPU 或 CPU
CUDA_DEVICE = None # "cuda:0" if GPU available, else None
# ================================
# 1. 文件切分与持久化
# ================================
def ingest_file(file_path: str) -> List[str]:
path = Path(file_path)
if not path.exists():
raise FileNotFoundError(f"文件未找到: {file_path}")
text = path.read_text(encoding="utf-8")
splitter = RecursiveCharacterTextSplitter(
chunk_size=CHUNK_SIZE,
chunk_overlap=CHUNK_OVERLAP
)
chunks = splitter.split_text(text)
paths = []
for chunk in chunks:
p = PERSIST_DIR / f"{uuid.uuid4().hex}.txt"
p.write_text(chunk.strip(), encoding="utf-8")
paths.append(str(p))
return paths
# ================================
# 2. 路径感知向量库（节省内存）
# ================================
class PathAwareStore:
def __init__(self):
print("🧠 初始化嵌入模型…")
self.embedder = pipeline(
task=Tasks.feature_extraction,
model=EMBEDDING_MODEL_ID,
device=CUDA_DEVICE
)
def _get_embedding(self, text: str) -> List[float]:
try:
result = self.embedder(input=text)
return result[OutputKeys.SENTENCE_EMBEDDINGS][0]
except Exception as e:
print(f"⚠️ 嵌入失败: {e}")
return [0.0] * 384 # bge-small 输出 384 维
def embed_query(self, text: str) -> List[float]:
return self._get_embedding(text)
def embed_documents(self, texts: List[str]) -> List[List[float]]:
return [self._get_embedding(t) for t in texts]
def add_paths(self, paths: List[str]):
docs = []
for p in paths:
try:
content = Path(p).read_text(encoding="utf-8")[:150]
docs.append(Document(page_content=content, metadata={"path": p}))
except Exception as e:
print(f"跳过文件 {p}: {e}")
if not docs:
print("❌ 无有效文档")
return
print("📊 构建向量数据库…")
db = FAISS.from_documents(docs, embedding=self)
db.save_local(str(PERSIST_DIR / "faiss_index"))
print("✅ 向量库已保存")
def similarity_search(self, query: str, k: int = TOP_K) -> List[Document]:
try:
db = FAISS.load_local(
str(PERSIST_DIR / "faiss_index"),
embeddings=self,
allow_dangerous_deserialization=True
)
docs = db.similarity_search(query, k=k)
for d in docs:
d.page_content = Path(d.metadata["path"]).read_text(encoding="utf-8")
return docs
except Exception as e:
print(f"❌ 检索失败: {e}")
return []
# ================================
# 3. 摘要压缩器
# ================================
class Compressor:
def __init__(self):
print("💬 初始化 Qwen 模型…")
self.generator = pipeline(
task=Tasks.text_generation,
model=LLM_MODEL_ID,
device=CUDA_DEVICE
)
def compress(self, docs: List[Document]) -> List[Document]:
out = []
for d in docs:
prompt = f"请用30字内中文摘要：\\n{d.page_content}"
try:
resp = self.generator(input=prompt, max_length=100, temperature=0.1)
summary = resp["text"].strip()
except Exception:
summary = "摘要失败"
out.append(Document(page_content=summary, metadata={"path": d.metadata["path"]}))
return out
def expand(self, doc: Document) -> Document:
raw = Path(doc.metadata["path"]).read_text(encoding="utf-8")
return Document(page_content=raw, metadata=doc.metadata)
# ================================
# 4. 智能体主类
# ================================
class LongContextAgent:
def __init__(self):
self.store = PathAwareStore()
self.compressor = Compressor()
self.memory = ConversationBufferWindowMemory(k=3)
def ingest(self, file_path: str):
print(f"📥 正在处理文件: {file_path}")
paths = ingest_file(file_path)
self.store.add_paths(paths)
print(f"✅ 成功索引 {len(paths)} 个段落")
def ask(self, query: str, expand_top: int = 1) -> str:
print(f"🔍 检索相关段落…")
docs = self.store.similarity_search(query)
if not docs:
return "未找到相关上下文。"
summaries = self.compressor.compress(docs)
expanded = [self.compressor.expand(summaries[i]) for i in range(expand_top)]
final_docs = expanded + summaries[expand_top:]
context = "\\n\\n".join(d.page_content for d in final_docs)
prompt = f"根据以下内容回答问题：\\n{context}\\n\\n问题：{query}\\n回答："
print("🧠 生成回答…")
try:
resp = self.compressor.generator(
input=prompt,
max_length=512,
temperature=0.5,
top_p=0.9,
do_sample=True
)
answer = resp["text"].strip().replace(prompt, "")
except Exception as e:
answer = f"回答生成失败: {str(e)}"
self.memory.save_context({"input": query}, {"output": answer})
return answer
# ================================
# 5. 主程序入口
# ================================
if __name__ == "__main__":
# 创建测试文件
test_file = "long_novel.txt"
if not Path(test_file).exists():
with open(test_file, "w", encoding="utf-8") as f:
f.write("""
主角李明来自山村，家境贫寒。他从小立志走出大山。
父亲病逝后，母亲希望他留下种地，但他坚持去城市打工求学。
经过十年努力，他成为工程师，返乡建设家乡，修路建校。
他说：“我答应过母亲，一定会回来。” 村民都很感激他。
""")
print(f"✅ 测试文件 '{test_file}' 已创建。")
# 初始化智能体
agent = LongContextAgent()
agent.ingest(test_file)
# 提问测试
questions = [
"主角叫什么名字？",
"他为什么离开故乡？",
"他后来为家乡做了什么？"
]
for q in questions:
print(f"\\n❓ 提问: {q}")
ans = agent.ask(q, expand_top=1)
print(f"✅ 回答: {ans}")
运行代码，输出如下：
✅ 测试文件 'long_novel.txt' 已创建。
🚀 初始化智能体…
🧠 初始化嵌入模型…
💬 初始化 Qwen 模型…
📥 正在处理文件: long_novel.txt
📊 构建向量数据库…
✅ 向量库已保存
✅ 成功索引 3 个段落
❓ 提问: 主角叫什么名字？
🔍 检索相关段落…
🧠 生成回答…
✅ 回答: 主角叫李明。
❓ 提问: 他为什么离开故乡？
🔍 检索相关段落…
🧠 生成回答…
✅ 回答: 他为了实现走出大山的梦想，坚持去城市打工求学，所以离开故乡。
❓ 提问: 他后来为家乡做了什么？
🔍 检索相关段落…
🧠 生成回答…
✅ 回答: 他成为工程师后返乡，为家乡修路建校，改善基础设施。
