人工智能

引言 多模态大模型（MLLM）能够同时理解和生成文本、图像、音频等多种模态的数据，开启了AI应用的新篇章。本文将深入探讨多模态LLM的技术原理、应用场景和开发实践。 多模态LLM基础 多模态架构 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 # 多模态模型架构对比 """ 1. Encoder-Decoder架构（如BLIP） - Image Encoder: ViT/Llama - Text Decoder: GPT-2 - Cross-attention连接 2. Encoder-only架构（如CLIP） - Image Encoder: ViT - Text Encoder: BERT - 对比学习 3. Decoder-only架构（如GPT-4V） - 统一的Transformer解码器 - 多模态输入投影 - 端到端生成 4. 项目架构（如LLaVA） - 预训练的视觉编码器 - 预训练的语言模型 - 简单的连接层训练 """ from transformers import AutoProcessor, LlavaForConditionalGeneration # 加载LLaVA模型 model = LlavaForConditionalGeneration.from_pretrained( "llava-hf/llava-1.5-7b-hf" ) processor = AutoProcessor.from_pretrained( "llava-hf/llava-1.5-7b-hf" ) 图像理解 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 import torch from PIL import Image from transformers import AutoModelForCausalLM, AutoProcessor class ImageUnderstanding: """图像理解助手""" def __init__(self, model_name="llava-hf/llava-1.5-7b-hf"): self.model = AutoModelForCausalLM.from_pretrained( model_name, torch_dtype=torch.float16, device_map="auto" ) self.processor = AutoProcessor.from_pretrained(model_name) def analyze_image( self, image_path: str, question: str ) -> str: """分析图像内容""" # 加载图像 image = Image.open(image_path).convert("RGB") # 准备输入 prompt = f"USER: <image>\n{question}\nASSISTANT:" inputs = self.processor( text=prompt, images=image, return_tensors="pt" ).to(self.model.device) # 生成响应 with torch.no_grad(): outputs = self.model.generate( **inputs, max_new_tokens=500, do_sample=False, ) # 解码响应 response = self.processor.decode( outputs[0], skip_special_tokens=True ) return response.split("ASSISTANT:")[-1].strip() def describe_scene(self, image_path: str) -> str: """描述场景""" return self.analyze_image( image_path, "Please describe this image in detail, including objects, people, activities, and the environment." ) def extract_text(self, image_path: str) -> str: """提取文字""" return self.analyze_image( image_path, "Extract all text visible in this image. Organize it logically." ) def detect_objects(self, image_path: str) -> str: """检测物体""" return self.analyze_image( image_path, "List all objects visible in this image with their locations and relationships." ) # 使用示例 assistant = ImageUnderstanding() # 分析图片 description = assistant.describe_scene("path/to/image.jpg") print(description) # 提取文字 text = assistant.extract_text("document.jpg") print(text) 图像+文本RAG 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 from typing import List import clip import torch from PIL import Image class MultimodalRAG: """多模态RAG系统""" def __init__(self): # 加载CLIP模型 self.device = "cuda" if torch.cuda.is_available() else "cpu" self.model, self.preprocess = clip.load("ViT-B/32", device=self.device) # 初始化向量存储 self.image_store = [] self.text_store = [] def add_documents( self, images: List[str] = None, texts: List[str] = None ): """添加多模态文档""" # 处理图像 if images: for img_path in images: image = Image.open(img_path) image_input = self.preprocess(image).unsqueeze(0).to(self.device) with torch.no_grad(): image_features = self.model.encode_image(image_input) self.image_store.append({ "path": img_path, "features": image_features.cpu(), "metadata": {"type": "image"} }) # 处理文本 if texts: for text in texts: text_input = clip.tokenize([text]).to(self.device) with torch.no_grad(): text_features = self.model.encode_text(text_input) self.text_store.append({ "content": text, "features": text_features.cpu(), "metadata": {"type": "text"} }) def retrieve( self, query_image: str = None, query_text: str = None, top_k: int = 5 ) -> List[dict]: """跨模态检索""" query_features = None # 图像查询 if query_image: image = Image.open(query_image) image_input = self.preprocess(image).unsqueeze(0).to(self.device) with torch.no_grad(): query_features = self.model.encode_image(image_input) # 文本查询 elif query_text: text_input = clip.tokenize([query_text]).to(self.device) with torch.no_grad(): query_features = self.model.encode_text(text_input) # 计算相似度 results = [] # 检索图像 for doc in self.image_store: similarity = torch.cosine_similarity( query_features, doc["features"].to(self.device) ).item() results.append({ "content": doc["path"], "score": similarity, "type": "image" }) # 检索文本 for doc in self.text_store: similarity = torch.cosine_similarity( query_features, doc["features"].to(self.device) ).item() results.append({ "content": doc["content"], "score": similarity, "type": "text" }) # 排序并返回top-k results.sort(key=lambda x: x["score"], reverse=True) return results[:top_k] def multimodal_rag( self, query_image: str, query_text: str ) -> str: """多模态RAG问答""" # 检索相关内容 image_results = self.retrieve(query_image=query_image) text_results = self.retrieve(query_text=query_text) # 构建prompt context = "Retrieved Information:\n" context += "\nRelevant Images:\n" for result in image_results[:3]: context += f"- {result['content']}\n" context += "\nRelevant Texts:\n" for result in text_results[:3]: context += f"- {result['content']}\n" prompt = f""" {context} Based on the above retrieved information and the provided image, answer the following question: {query_text} """ # 使用多模态LLM生成答案 response = self.analyze_image(query_image, prompt) return response 视频理解 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 import cv2 import numpy as np from typing import List class VideoAnalyzer: """视频分析器""" def __init__(self, mlm_model): self.mlm_model = mlm_model def extract_key_frames( self, video_path: str, num_frames: int = 10 ) -> List[str]: """提取关键帧""" cap = cv2.VideoCapture(video_path) total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) frame_indices = np.linspace(0, total_frames - 1, num_frames, dtype=int) key_frames = [] for idx in frame_indices: cap.set(cv2.CAP_PROP_POS_FRAMES, idx) ret, frame = cap.read() if ret: frame_path = f"frame_{idx}.jpg" cv2.imwrite(frame_path, frame) key_frames.append(frame_path) cap.release() return key_frames def analyze_video( self, video_path: str, question: str ) -> str: """分析视频内容""" # 提取关键帧 key_frames = self.extract_key_frames(video_path) # 分析关键帧 frame_descriptions = [] for frame_path in key_frames: description = self.mlm_model.describe_scene(frame_path) frame_descriptions.append(description) # 综合分析 prompt = f""" Here are descriptions of key frames from a video: {chr(10).join([f'Frame {i+1}: {desc}' for i, desc in enumerate(frame_descriptions)])} Based on these frame descriptions, answer: {question} """ # 使用文本LLM生成答案 response = self.text_llm.generate(prompt) return response def detect_actions(self, video_path: str) -> List[str]: """检测动作""" key_frames = self.extract_key_frames(video_path, num_frames=20) actions = [] for i, frame_path in enumerate(key_frames): action = self.mlm_model.analyze_image( frame_path, "What action is being performed in this frame? Be concise." ) actions.append(f"Frame {i+1}: {action}") return actions def summarize_video(self, video_path: str) -> str: """视频摘要""" # 提取关键帧 key_frames = self.extract_key_frames(video_path) # 生成摘要 prompt = "Create a concise summary of this video based on these key frames:" for i, frame_path in enumerate(key_frames[:5]): frame_desc = self.mlm_model.describe_scene(frame_path) prompt += f"\nFrame {i+1}: {frame_desc}" summary = self.mlm_model.text_llm.generate(prompt) return summary 音频处理 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 import whisper import torch from typing import Dict class MultimodalAudioAssistant: """多模态音频助手""" def __init__(self): # 加载Whisper模型 self.audio_model = whisper.load_model("base") # 加载多模态LLM self.mlm_model = AutoModelForCausalLM.from_pretrained( "llava-hf/llava-1.5-7b-hf" ) def transcribe_audio( self, audio_path: str ) -> Dict[str, any]: """转录音频""" # 转录 result = self.audio_model.transcribe( audio_path, language="zh", task="transcribe" ) return { "text": result["text"], "language": result["language"], "segments": result["segments"] } def audio_question_answering( self, audio_path: str, question: str ) -> str: """音频问答""" # 转录音频 transcription = self.transcribe_audio(audio_path) audio_text = transcription["text"] # 使用LLM回答问题 prompt = f""" Audio Transcript: {audio_text} Question: {question} Based on the audio transcript, provide a detailed answer. """ response = self.mlm_model.generate(prompt) return response def audio_visual_sync( self, audio_path: str, video_path: str ) -> Dict[str, any]: """音视频同步分析""" # 转录音频 audio_text = self.transcribe_audio(audio_path) # 提取视频关键帧 key_frames = self.extract_key_frames(video_path) # 分析音视频关联 prompt = f""" Audio: {audio_text['text']} Visual Content: {[self.mlm_model.describe_scene(frame) for frame in key_frames[:3]]} Analyze the relationship between the audio and visual content. Are they consistent? What is the overall message? """ analysis = self.mlm_model.generate(prompt) return { "audio_text": audio_text, "visual_summary": analysis } 实战应用 应用1：多模态文档分析 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 from transformers import DonutProcessor, VisionEncoderDecoderModel class DocumentAnalyzer: """文档分析器""" def __init__(self): # 加载Donut模型 self.processor = DonutProcessor.from_pretrained( "naver-clova-ix/donut-base-finetuned-docvqa" ) self.model = VisionEncoderDecoderModel.from_pretrained( "naver-clova-ix/donut-base-finetuned-docvqa" ) def analyze_document( self, image_path: str, question: str ) -> str: """分析文档图像""" # 加载文档图像 image = Image.open(image_path).convert("RGB") # 准备输入 prompt = f"<s>{question}</s>" task_prompt = f"<s_docvqa><s_question>{question}</s_answer><s>" inputs = self.processor( image, task_prompt, return_tensors="pt" ) # 生成答案 with torch.no_grad(): outputs = self.model.generate( inputs.pixel_values, inputs.input_ids[0:1], ) # 解码 generated_text = self.processor.batch_decode(outputs)[0] # 提取答案 answer = generated_text.split("</s_answer>")[-1].strip() return answer def extract_table(self, image_path: str) -> str: """提取表格""" return self.analyze_document( image_path, "Extract all tables from this document in Markdown format." ) def extract_form_fields(self, image_path: str) -> Dict[str, str]: """提取表单字段""" result = self.analyze_document( image_path, "List all form fields in this document with their labels and values." ) # 解析结果为结构化数据 fields = {} for line in result.split("\n"): if ":" in line: label, value = line.split(":", 1) fields[label.strip()] = value.strip() return fields 应用2：多模态聊天机器人 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 from fastapi import FastAPI, UploadFile, File from pydantic import BaseModel from typing import Optional app = FastAPI() class MultimodalChatBot: """多模态聊天机器人""" def __init__(self): self.vlm_model = LlavaForConditionalGeneration.from_pretrained( "llava-hf/llava-1.5-7b-hf", torch_dtype=torch.float16, device_map="auto" ) self.processor = AutoProcessor.from_pretrained( "llava-hf/llava-1.5-7b-hf" ) # 对话历史 self.conversation_history = {} async def chat( self, user_id: str, message: str, image: Optional[UploadFile] = None ) -> str: """多模态对话""" # 获取历史 history = self.conversation_history.get(user_id, []) # 准备输入 if image: # 有图像 image_bytes = await image.read() image_pil = Image.open(io.BytesIO(image_bytes)).convert("RGB") prompt = self._build_prompt_with_image(history, message) inputs = self.processor( text=prompt, images=image_pil, return_tensors="pt" ).to(self.vlm_model.device) # 生成 with torch.no_grad(): outputs = self.vlm_model.generate( **inputs, max_new_tokens=500, do_sample=True, temperature=0.7, ) response = self.processor.decode(outputs[0], skip_special_tokens=True) else: # 纯文本 prompt = self._build_prompt(history, message) response = self.text_llm.generate(prompt) # 更新历史 history.append({"role": "user", "content": message}) history.append({"role": "assistant", "content": response}) self.conversation_history[user_id] = history[-10:] # 保留最近10轮 return response def _build_prompt_with_image(self, history, message): prompt = "USER: <image>\n" for h in history: prompt += f"{h['role'].upper()}: {h['content']}\n" prompt += f"USER: {message}\nASSISTANT:" return prompt def _build_prompt(self, history, message): prompt = "" for h in history: prompt += f"{h['role'].upper()}: {h['content']}\n" prompt += f"USER: {message}\nASSISTANT:" return prompt chatbot = MultimodalChatBot() @app.post("/chat/{user_id}") async def chat_endpoint( user_id: str, message: str = Form(...), image: UploadFile = File(None) ): response = await chatbot.chat(user_id, message, image) return {"response": response} 总结 多模态大模型正在快速演进，从单一的文本理解发展到图像、视频、音频的综合理解。 ...

引言 大模型微调是将预训练模型适配到特定任务的关键技术。从传统的全量微调到参数高效的LoRA、QLoRA，微调技术不断发展。本文将深入探讨各种微调方法的原理、实现和最佳实践。 微调基础概念 为什么需要微调 1 2 3 4 5 6 7 8 9 10 11 # 预训练模型 vs 微调模型 # 预训练模型（通用） pretrained_model = "gpt-4" response = pretrained_model.generate("解释什么是量子纠缠") # 输出：通用的、百科全书式的解释 # 微调模型（特定领域） fine_tuned_model = "gpt-4-quantum-physics" # 经过量子物理领域微调 response = fine_tuned_model.generate("解释什么是量子纠缠") # 输出：更专业、更准确、使用领域术语的解释 微调类型 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 // 微调类型对比 interface FineTuningTypes { full: { name: "全量微调", description: "更新所有模型参数", pros: ["效果最好", "适配最完整"], cons: ["成本高", "需要大量数据", "存储需求大"] }, partial: { name: "部分微调", description: "只更新部分层", pros: ["平衡效果和成本"], cons: ["需要经验确定哪些层"] }, peft: { name: "参数高效微调", description: "只更新少量参数", pros: ["成本极低", "速度快", "存储小"], cons: ["效果略低于全量微调"] } } 全量微调 基础实现 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 import torch from transformers import ( AutoModelForCausalLM, AutoTokenizer, TrainingArguments, Trainer, DataCollatorForLanguageModeling ) from datasets import load_dataset # 加载预训练模型和分词器 model_name = "bert-base-chinese" model = AutoModelForCausalLM.from_pretrained(model_name) tokenizer = AutoTokenizer.from_pretrained(model_name) # 准备数据集 dataset = load_dataset("csv", data_files="training_data.csv") def tokenize_function(examples): return tokenizer( examples["text"], padding="max_length", truncation=True, max_length=512 ) tokenized_datasets = dataset.map(tokenize_function, batched=True) # 训练参数 training_args = TrainingArguments( output_dir="./results", evaluation_strategy="epoch", learning_rate=2e-5, per_device_train_batch_size=8, per_device_eval_batch_size=8, num_train_epochs=3, weight_decay=0.01, save_strategy="epoch", load_best_model_at_end=True, ) # 数据整理器 data_collator = DataCollatorForLanguageModeling( tokenizer=tokenizer, mlm=False, # Causal LM不需要MLM ) # 创建Trainer trainer = Trainer( model=model, args=training_args, train_dataset=tokenized_datasets["train"], eval_dataset=tokenized_datasets["validation"], data_collator=data_collator, ) # 开始训练 trainer.train() # 保存模型 trainer.save_model("./my_finetuned_model") tokenizer.save_pretrained("./my_finetuned_model") 自定义训练循环 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 from torch.utils.data import DataLoader from tqdm import tqdm # 创建DataLoader train_dataloader = DataLoader( tokenized_datasets["train"], shuffle=True, batch_size=8, collate_fn=data_collator ) eval_dataloader = DataLoader( tokenized_datasets["validation"], batch_size=8, collate_fn=data_collator ) # 优化器 optimizer = torch.optim.AdamW(model.parameters(), lr=2e-5) # 学习率调度器 num_training_steps = len(train_dataloader) * 3 lr_scheduler = torch.optim.lr_scheduler.CosineAnnealingLR( optimizer, T_max=num_training_steps ) # 训练循环 device = torch.device("cuda" if torch.cuda.is_available() else "cpu") model.to(device) for epoch in range(3): model.train() total_loss = 0 progress_bar = tqdm(train_dataloader, desc=f"Epoch {epoch+1}") for batch in progress_bar: batch = {k: v.to(device) for k, v in batch.items()} # 前向传播 outputs = model(**batch) loss = outputs.loss # 反向传播 loss.backward() optimizer.step() lr_scheduler.step() optimizer.zero_grad() total_loss += loss.item() progress_bar.set_postfix({"loss": loss.item()}) avg_train_loss = total_loss / len(train_dataloader) # 评估 model.eval() eval_loss = 0 with torch.no_grad(): for batch in eval_dataloader: batch = {k: v.to(device) for k, v in batch.items()} outputs = model(**batch) eval_loss += outputs.loss.item() avg_eval_loss = eval_loss / len(eval_dataloader) print(f"Epoch {epoch+1}: Train Loss = {avg_train_loss:.4f}, Eval Loss = {avg_eval_loss:.4f}") LoRA微调 LoRA原理 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 # LoRA (Low-Rank Adaptation) # 核心思想：在权重矩阵旁添加低秩分解矩阵 import torch import torch.nn as nn class LoRALayer(nn.Module): """LoRA层""" def __init__( self, original_layer: nn.Linear, rank: int = 8, alpha: float = 32 ): super().__init__() self.original_layer = original_layer self.rank = rank self.alpha = alpha # 获取原始层维度 in_features = original_layer.in_features out_features = original_layer.out_features # LoRA参数（低秩矩阵） self.lora_A = nn.Parameter(torch.zeros(rank, in_features)) self.lora_B = nn.Parameter(torch.zeros(out_features, rank)) # 初始化 nn.init.kaiming_uniform_(self.lora_A, a=math.sqrt(5)) nn.init.zeros_(self.lora_B) self.scaling = alpha / rank def forward(self, x): # 原始层输出 original_output = self.original_layer(x) # LoRA输出: B * A * x lora_output = (x @ self.lora_A.T @ self.lora_B.T) * self.scaling return original_output + lora_output # 使用示例 original_linear = nn.Linear(768, 768) lora_linear = LoRALayer(original_linear, rank=8) # 参数量对比 original_params = sum(p.numel() for p in original_linear.parameters()) lora_params = sum(p.numel() for p in lora_linear.parameters()[ 'lora_A', 'lora_B' ]) print(f"原始参数: {original_params:,}") print(f"LoRA参数: {lora_params:,}") print(f"参数减少: {(1 - lora_params / original_params) * 100:.2f}%") 使用PEFT库 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 from peft import LoraConfig, get_peft_model, TaskType from transformers import AutoModelForCausalLM # 加载基础模型 model = AutoModelForCausalLM.from_pretrained( "bigscience/bloom-7b1", torch_dtype=torch.float16, device_map="auto" ) # LoRA配置 lora_config = LoraConfig( task_type=TaskType.CAUSAL_LM, # 任务类型 inference_mode=False, # 训练模式 r=8, # LoRA rank lora_alpha=32, # LoRA alpha lora_dropout=0.1, # Dropout target_modules=["q_proj", "v_proj"], # 要应用LoRA的模块 ) # 应用LoRA model = get_peft_model(model, lora_config) # 查看可训练参数 model.print_trainable_parameters() # 输出类似： # trainable params: 2,621,440 || all params: 7,111,635,456 || trainable%: 0.0368% # 训练（与普通模型相同） trainer = Trainer( model=model, args=training_args, train_dataset=train_dataset, eval_dataset=eval_dataset, ) trainer.train() # 保存LoRA权重 model.save_pretrained("./my_lora_model") # 加载LoRA权重 from peft import PeftModel base_model = AutoModelForCausalLM.from_pretrained("bigscience/bloom-7b1") model = PeftModel.from_pretrained(base_model, "./my_lora_model") 不同LoRA策略 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 # 策略1：只微调Attention层 lora_config_attention = LoraConfig( task_type=TaskType.CAUSAL_LM, r=16, lora_alpha=32, target_modules=["q_proj", "k_proj", "v_proj", "o_proj"], ) # 策略2：微调Attention和MLP层 lora_config_full = LoraConfig( task_type=TaskType.CAUSAL_LM, r=16, lora_alpha=32, target_modules=[ "q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj" ], ) # 策略3：所有Linear层 lora_config_all = LoraConfig( task_type=TaskType.CAUSAL_LM, r=8, lora_alpha=16, target_modules=[".*"], # 使用正则匹配所有 ) # 策略4：不同层使用不同rank from peft import LoraConfig class MultiRankLoraConfig(LoraConfig): """多rank LoRA配置""" def __init__( self, layer_ranks: dict, # {"layer_name": rank} **kwargs ): super().__init__(**kwargs) self.layer_ranks = layer_ranks # 使用 config = MultiRankLoraConfig( task_type=TaskType.CAUSAL_LM, layer_ranks={ "model.layers.0": 4, "model.layers.10": 8, "model.layers.20": 16, } ) QLoRA微调 QLoRA原理 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 # QLoRA (Quantized LoRA) # 核心：量化+LoRA，在保持性能的同时大幅减少显存 import torch from transformers import BitsAndBytesConfig # 量化配置 bnb_config = BitsAndBytesConfig( load_in_4bit=True, # 4-bit量化 bnb_4bit_compute_dtype=torch.float16, bnb_4bit_use_double_quant=True, # 双重量化 bnb_4bit_quant_type="nf4", # NF4量化类型 ) # 加载量化模型 model = AutoModelForCausalLM.from_pretrained( "meta-llama/Llama-2-7b-hf", quantization_config=bnb_config, device_map="auto" ) # 应用LoRA from peft import LoraConfig, get_peft_model lora_config = LoraConfig( r=16, lora_alpha=32, target_modules=["q_proj", "v_proj"], lora_dropout=0.05, bias="none", task_type=TaskType.CAUSAL_LM, ) model = get_peft_model(model, lora_config) # 训练 trainer = Trainer( model=model, args=TrainingArguments( output_dir="./qlora_output", per_device_train_batch_size=4, gradient_accumulation_steps=4, learning_rate=2e-4, logging_steps=10, save_steps=100, num_train_epochs=3, ), train_dataset=train_dataset, ) trainer.train() # 显存使用对比（7B模型） """ 全量微调: - 模型权重: ~28GB (FP16) - 梯度: ~28GB - 优化器状态: ~84GB - 总计: ~140GB (需要8x A100 80GB) LoRA微调: - 模型权重: ~28GB - LoRA参数: ~100MB - 梯度: ~100MB - 优化器状态: ~300MB - 总计: ~28.5GB (1x A100 40GB) QLoRA微调: - 量化权重: ~7GB (4-bit) - LoRA参数: ~100MB - 梯度: ~100MB - 优化器状态: ~300MB - 总计: ~7.5GB (1x RTX 3090 24GB) """ 完整QLoRA训练流程 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 import torch from transformers import ( AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig, TrainingArguments, Trainer, ) from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training from datasets import load_dataset from transformers import TrainerCallback # 1. 加载模型（4-bit量化） bnb_config = BitsAndBytesConfig( load_in_4bit=True, bnb_4bit_quant_type="nf4", bnb_4bit_compute_dtype=torch.bfloat16, bnb_4bit_use_double_quant=True, ) model = AutoModelForCausalLM.from_pretrained( "meta-llama/Llama-2-7b-hf", quantization_config=bnb_config, device_map="auto", trust_remote_code=True, ) tokenizer = AutoTokenizer.from_pretrained( "meta-llama/Llama-2-7b-hf", trust_remote_code=True, ) tokenizer.pad_token = tokenizer.eos_token # 2. 准备模型进行训练 model = prepare_model_for_kbit_training(model) # 3. 配置LoRA lora_config = LoraConfig( r=16, lora_alpha=32, target_modules=["q_proj", "k_proj", "v_proj", "o_proj"], lora_dropout=0.05, bias="none", task_type=TaskType.CAUSAL_LM, ) model = get_peft_model(model, lora_config) model.print_trainable_parameters() # 4. 准备数据 dataset = load_dataset("json", data_files="training_data.json") def format_prompt(example): return { "text": f"### Instruction:\n{example['instruction']}\n\n### Response:\n{example['output']}" } dataset = dataset.map(format_prompt) def tokenize_function(examples): return tokenizer( examples["text"], padding="max_length", truncation=True, max_length=512, ) tokenized_dataset = dataset.map(tokenize_function, batched=True) # 5. 训练参数 training_args = TrainingArguments( output_dir="./qlora_checkpoints", num_train_epochs=3, per_device_train_batch_size=4, gradient_accumulation_steps=4, warmup_steps=100, learning_rate=2e-4, fp16=True, logging_steps=10, optim="paged_adamw_8bit", # 分页优化器 save_strategy="steps", save_steps=100, evaluation_strategy="steps", eval_steps=100, load_best_model_at_end=True, report_to=["wandb"], # 或 "tensorboard" run_name="qlora-finetune", ) # 6. 自定义回调 class LoggingCallback(TrainerCallback): def on_log(self, args, state, control, logs=None, **kwargs): if logs: print(f"Step: {state.global_step}, Loss: {logs.get('loss', 'N/A')}") # 7. 训练 trainer = Trainer( model=model, args=training_args, train_dataset=tokenized_dataset["train"], eval_dataset=tokenized_dataset["validation"], callbacks=[LoggingCallback()], ) trainer.train() # 8. 保存 model.save_pretrained("./final_qlora_model") tokenizer.save_pretrained("./final_qlora_model") # 9. 合并权重（可选） merged_model = model.merge_and_unload() merged_model.save_pretrained("./merged_model") 其他PEFT方法 Prefix Tuning 1 2 3 4 5 6 7 8 9 10 11 12 13 14 from peft import PrefixTuningConfig, get_peft_model # Prefix Tuning配置 prefix_config = PrefixTuningConfig( task_type=TaskType.CAUSAL_LM, num_virtual_tokens=20, # 虚拟token数量 prefix_projection=True, # 使用投影层 ) model = get_peft_model(model, prefix_config) # Prefix Tuning在每一层添加可训练的前缀向量 # 参数量：num_layers * num_virtual_tokens * hidden_dim # 对于Llama-2-7b: 32 * 20 * 4096 ≈ 2.6M 参数 Prompt Tuning 1 2 3 4 5 6 7 8 9 10 11 12 13 14 from peft import PromptTuningConfig, get_peft_model # Prompt Tuning配置 prompt_config = PromptTuningConfig( task_type=TaskType.CAUSAL_LM, prompt_tuning_init="TEXT", # 使用文本初始化 prompt_tuning_init_text="分类以下文本：", num_virtual_tokens=8, ) model = get_peft_model(model, prompt_config) # Prompt Tuning只在输入层添加可训练的prompt # 参数量最少，但效果相对较弱 Adapter 1 2 3 4 5 6 7 8 9 10 11 12 from transformers import AdapterType, BertAdapterModel # 加载带Adapter的模型 model = BertAdapterModel.from_pretrained("bert-base-uncased") # 添加Adapter model.add_adapter("sentiment", AdapterType.text_task) model.train_adapter("sentiment") # Adapter在每个Transformer层后添加小型 bottleneck 层 # 结构: down_proj -> nonlinearity -> up_proj # 参数量更少，但性能通常不如LoRA 数据准备 指令微调数据 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 # Alpaca格式 alpaca_dataset = [ { "instruction": "解释什么是机器学习？", "input": "", "output": "机器学习是人工智能的一个分支..." }, { "instruction": "翻译以下句子", "input": "Hello, world!", "output": "你好，世界！" } ] # 转换为训练格式 def format_alpaca(example): if example["input"]: prompt = f"### Instruction:\n{example['instruction']}\n\n### Input:\n{example['input']}\n\n### Response:\n{example['output']}" else: prompt = f"### Instruction:\n{example['instruction']}\n\n### Response:\n{example['output']}" return {"text": prompt} # ShareGPT格式（多轮对话） sharegpt_dataset = [ { "conversations": [ {"from": "human", "value": "你好"}, {"from": "gpt", "value": "你好！有什么可以帮你的？"}, {"from": "human", "value": "介绍一下Python"}, {"from": "gpt", "value": "Python是一种高级编程语言..."} ] } ] def format_sharegpt(example): conversations = example["conversations"] # 构建对话历史 formatted = [] for conv in conversations: role = "User" if conv["from"] == "human" else "Assistant" formatted.append(f"{role}: {conv['value']}") return {"text": "\n".join(formatted)} 数据增强 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 import random import nlpaug.augmenter.word as naw class DataAugmenter: """数据增强器""" def __init__(self): # 同义词替换 self.synonym_aug = naw.SynonymAug(aug_src='wordnet') # 随机删除 self.delete_aug = naw.RandomWordAug(action="delete") # 随机交换 self.swap_aug = naw.RandomWordAug(action="swap") def augment(self, text: str, num_augmented: int = 3) -> list: """增强数据""" augmented_texts = [text] for _ in range(num_augmented): aug_type = random.choice(["synonym", "delete", "swap"]) if aug_type == "synonym": aug_text = self.synonym_aug.augment(text) elif aug_type == "delete": aug_text = self.delete_aug.augment(text) else: aug_text = self.swap_aug.augment(text) augmented_texts.append(aug_text[0] if isinstance(aug_text, list) else aug_text) return augmented_texts def paraphrase(self, text: str, llm) -> str: """使用LLM改写""" prompt = f"请改写以下文本，保持原意但使用不同的表达：\n{text}" return llm.generate(prompt) # 使用 augmenter = DataAugmenter() original = "机器学习是人工智能的重要分支" augmented = augmenter.augment(original, num_augmented=3) # [ # "机器学习是人工智能的重要分支", # "机器学习是AI的关键组成部分", # "ML是人工智能的核心领域", # "机器学习属于人工智能范畴" # ] 评估和测试 Perplexity评估 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 import torch from tqdm import tqdm def calculate_perplexity(model, dataloader, device): """计算困惑度""" model.eval() total_loss = 0 total_tokens = 0 with torch.no_grad(): for batch in tqdm(dataloader, desc="Evaluating"): batch = {k: v.to(device) for k, v in batch.items()} outputs = model(**batch, labels=batch["input_ids"]) loss = outputs.loss total_loss += loss.item() * batch["input_ids"].numel() total_tokens += batch["input_ids"].numel() avg_loss = total_loss / total_tokens perplexity = torch.exp(torch.tensor(avg_loss)) return perplexity.item() # 使用 perplexity = calculate_perplexity(model, eval_dataloader, device) print(f"Perplexity: {perplexity:.2f}") 任务特定评估 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 # 分类任务评估 from sklearn.metrics import accuracy_score, f1_score, classification_report def evaluate_classification(model, dataloader, device): """评估分类任务""" model.eval() predictions = [] true_labels = [] with torch.no_grad(): for batch in tqdm(dataloader): batch = {k: v.to(device) for k, v in batch.items()} outputs = model(**batch) preds = torch.argmax(outputs.logits, dim=-1) predictions.extend(preds.cpu().numpy()) true_labels.extend(batch["labels"].cpu().numpy()) accuracy = accuracy_score(true_labels, predictions) f1 = f1_score(true_labels, predictions, average="weighted") print(f"Accuracy: {accuracy:.4f}") print(f"F1 Score: {f1:.4f}") print("\nClassification Report:") print(classification_report(true_labels, predictions)) return accuracy, f1 # 生成任务评估 import evaluate bleu = evaluate.load("bleu") rouge = evaluate.load("rouge") def evaluate_generation(model, dataloader, tokenizer, device): """评估生成任务""" model.eval() predictions = [] references = [] with torch.no_grad(): for batch in tqdm(dataloader): batch = {k: v.to(device) for k, v in batch.items()} outputs = model.generate( **batch, max_new_tokens=100, do_sample=True, temperature=0.7 ) decoded_preds = tokenizer.batch_decode(outputs, skip_special_tokens=True) decoded_refs = tokenizer.batch_decode(batch["input_ids"], skip_special_tokens=True) predictions.extend(decoded_preds) references.extend(decoded_refs) # BLEU bleu_score = bleu.compute( predictions=predictions, references=[[ref] for ref in references] ) # ROUGE rouge_score = rouge.compute( predictions=predictions, references=references ) return { "bleu": bleu_score, "rouge": rouge_score } 实战案例 案例：医疗问答系统微调 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 from transformers import AutoModelForCausalLM, AutoTokenizer from peft import LoraConfig, get_peft_model # 1. 准备医疗数据 medical_data = [ { "instruction": "什么是高血压？", "output": "高血压是指血液在血管中流动时对血管壁产生的压力值持续高于正常值的疾病..." }, { "instruction": "糖尿病有哪些症状？", "output": "糖尿病的主要症状包括：多饮、多尿、多食、体重下降（三多一少）..." }, # ... 更多医疗问答 ] # 2. 加载模型（使用QLoRA节省显存） model_name = "meta-llama/Llama-2-7b-chat-hf" bnb_config = BitsAndBytesConfig( load_in_4bit=True, bnb_4bit_quant_type="nf4", bnb_4bit_compute_dtype=torch.bfloat16, ) model = AutoModelForCausalLM.from_pretrained( model_name, quantization_config=bnb_config, device_map="auto" ) tokenizer = AutoTokenizer.from_pretrained(model_name) # 3. 配置LoRA lora_config = LoraConfig( r=16, lora_alpha=32, target_modules=["q_proj", "v_proj"], lora_dropout=0.1, bias="none", task_type=TaskType.CAUSAL_LM, ) model = get_peft_model(model, lora_config) # 4. 训练 training_args = TrainingArguments( output_dir="./medical_llm", num_train_epochs=3, per_device_train_batch_size=4, gradient_accumulation_steps=4, learning_rate=1e-4, fp16=True, logging_steps=10, save_steps=100, ) trainer = Trainer( model=model, args=training_args, train_dataset=tokenized_medical_dataset, ) trainer.train() # 5. 测试 model.eval() prompt = "什么是高血压？" inputs = tokenizer(prompt, return_tensors="pt").to(model.device) with torch.no_grad(): outputs = model.generate( **inputs, max_new_tokens=200, temperature=0.7, do_sample=True ) response = tokenizer.decode(outputs[0], skip_special_tokens=True) print(response) 总结 大模型微调技术从全量微调发展到参数高效微调（PEFT），大幅降低了微调成本和门槛。 ...

引言 AI Agent的强大能力来自于其工作流编排能力——将复杂任务分解为多个步骤，并智能地协调执行。从简单的链式调用到复杂的多Agent协作，工作流编排是Agent系统的核心。本文将深入探讨主流的Agent编排框架和实战技巧。 LangChain Chains 基础Chain 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 from langchain.chains import LLMChain from langchain.prompts import PromptTemplate from langchain_openai import OpenAI # 创建LLM llm = OpenAI(temperature=0) # 创建Prompt模板 prompt_template = PromptTemplate( input_variables=["product"], template="为{product}写一段吸引人的产品描述。" ) # 创建Chain chain = LLMChain(llm=llm, prompt=prompt_template) # 运行 description = chain.run(product="智能手表") print(description) Sequential Chain 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 from langchain.chains import SequentialChain # Chain 1: 生成产品名称 name_chain = LLMChain( llm=llm, prompt=PromptTemplate( input_variables=["product_type"], template="为{product_type}产品想一个创意名称，只返回名称。" ), output_key="product_name" ) # Chain 2: 生成Slogan slogan_chain = LLMChain( llm=llm, prompt=PromptTemplate( input_variables=["product_name"], template="为{product_name}写一句简短有力的广告语。" ), output_key="slogan" ) # Chain 3: 生成完整描述 description_chain = LLMChain( llm=llm, prompt=PromptTemplate( input_variables=["product_name", "slogan"], template="产品名称：{product_name}\n广告语：{slogan}\n请基于以上信息写一段100字的产品描述。" ), output_key="description" ) # 组合Chain overall_chain = SequentialChain( chains=[name_chain, slogan_chain, description_chain], input_variables=["product_type"], output_variables=["product_name", "slogan", "description"] ) # 执行 result = overall_chain("智能手表") print(result) # { # 'product_name': 'TimePulse', # 'slogan': 'TimePulse - 让时间更有价值', # 'description': '...' # } Conditional Chain 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 from langchain.chains import TransformChain # 条件判断函数 def categorize_price(inputs: dict) -> dict: price = inputs.get('price', 0) if price < 100: category = "low" elif price < 500: category = "medium" else: category = "high" return {"price_category": category} # 条件Chain price_categorize_chain = TransformChain( transform=categorize_price, input_variables=["price"], output_variables=["price_category"] ) # 不同价格段的不同处理 low_price_chain = LLMChain( llm=llm, prompt=PromptTemplate( input_variables=["product"], template="{product}是经济实惠的选择，适合预算有限的用户。写一段强调性价比的描述。" ) ) high_price_chain = LLMChain( llm=llm, prompt=PromptTemplate( input_variables=["product"], template="{product}是高端产品，强调其品质和独特价值。写一段描述。" ) ) from langchain.chains import RouterChain # 路由Chain router_chain = RouterChain( chains={ "low": low_price_chain, "medium": medium_price_chain, "high": high_price_chain }, default_chain=medium_price_chain ) LCEL (LangChain Expression Language) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser # 使用LCEL构建Chain prompt = PromptTemplate.from_template( "Tell me a joke about {topic}" ) # 使用管道操作符 (|) chain = ( prompt | llm | StrOutputParser() ) # 等价于 chain = prompt | llm | StrOutputParser() # 执行 result = chain.invoke({"topic": "programming"}) print(result) # 复杂的LCEL示例 from langchain_community.utilities import WikipediaSearch wiki_search = WikipediaSearch() research_chain = ( { "context": lambda x: wiki_search.run(x["topic"]), "topic": RunnablePassthrough() } | PromptTemplate.from_template( "Topic: {topic}\n\nResearch: {context}\n\nBased on the research, explain {topic} in simple terms." ) | llm | StrOutputParser() ) LangChain Agents ReAct Agent 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 from langchain.agents import AgentExecutor, create_react_agent from langchain.tools import Tool from langchain_openai import OpenAI # 定义工具 def search_tool(query: str) -> str: """搜索工具""" # 实际应用中调用搜索API return f"搜索'{query}'的结果：..." def calculator_tool(expression: str) -> str: """计算器工具""" try: result = eval(expression) return f"计算结果：{result}" except: return "计算错误" tools = [ Tool( name="Search", func=search_tool, description="用于搜索网络信息，输入应该是搜索查询" ), Tool( name="Calculator", func=calculator_tool, description="用于数学计算，输入应该是数学表达式" ) ] # 创建Agent llm = OpenAI(temperature=0) prompt = PromptTemplate.from_template( """Answer the following questions as best you can. You have access to the following tools: {tools} Use the following format: Question: the input question you must answer Thought: you should always think about what to do Action: the action to take, should be one of [{tool_names}] Action Input: the input to the action Observation: the result of the action ... (this Thought/Action/Action Input/Observation can repeat N times) Thought: I now know the final answer Final Answer: the final answer to the original input question Begin! Question: {input} Thought: {agent_scratchpad}""" ) agent = create_react_agent( llm=llm, tools=tools, prompt=prompt ) # 创建Agent执行器 agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, max_iterations=5 ) # 执行 result = agent_executor.invoke({ "input": "苹果公司现在的股价是多少？如果我有100股，总价值多少？" }) print(result["output"]) Custom Agent 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 from langchain.agents import AgentExecutor, BaseSingleActionAgent from langchain_openai import BaseOpenAI class CustomAgent(BaseSingleActionAgent): """自定义Agent""" llm: BaseOpenAI tools: list[Tool] @property def input_keys(self): return ["input"] def plan( self, intermediate_steps: list[tuple[str, str]], **kwargs: Any ) -> tuple[AgentAction, str]: """规划下一步行动""" user_input = kwargs["input"] # 构建思考过程 thoughts = "" for action, observation in intermediate_steps: thoughts += f"Action: {action.tool}\n" thoughts += f"Input: {action.tool_input}\n" thoughts += f"Observation: {observation}\n" # 让LLM决定下一步 prompt = f""" 输入: {user_input} 之前的步骤: {thoughts} 可用工具: {[tool.name for tool in self.tools]} 请决定下一步行动，格式为： Action: [工具名称] Input: [工具输入] 或如果已完成： Final Answer: [最终答案] """ response = self.llm.predict(prompt) # 解析响应 if "Final Answer:" in response: final_answer = response.split("Final Answer:")[-1].strip() return AgentAction( tool="FINAL", tool_input=final_answer, log=response ), final_answer else: # 提取Action和Input action_line = [l for l in response.split("\n") if "Action:" in l][0] input_line = [l for l in response.split("\n") if "Input:" in l][0] tool_name = action_line.split("Action:")[-1].strip() tool_input = input_line.split("Input:")[-1].strip() return AgentAction( tool=tool_name, tool_input=tool_input, log=response ), "" async def aplan( self, intermediate_steps: list[tuple[str, str]], **kwargs: Any ) -> tuple[AgentAction, str]: """异步规划""" return self.plan(intermediate_steps, **kwargs) # 使用自定义Agent custom_agent = CustomAgent( llm=OpenAI(temperature=0), tools=tools ) agent_executor = AgentExecutor( agent=custom_agent, tools=tools ) result = agent_executor.invoke({"input": "查询北京今天的天气"}) AutoGPT模式 基础AutoGPT实现 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 from typing import List, Dict import json class AutoGPTAgent: """AutoGPT风格的Agent""" def __init__( self, name: str, role: str, goals: List[str], llm, tools: Dict[str, callable] ): self.name = name self.role = role self.goals = goals self.llm = llm self.tools = tools self.memory = [] self.task_list = [] def think(self) -> Dict: """思考下一步行动""" prompt = f""" Name: {self.name} Role: {self.role} Goals: {', '.join(self.goals)} Memory: {self.format_memory()} Current Tasks: {self.format_tasks()} 请决定下一步行动。返回JSON格式： {{ "thought": "思考过程", "reasoning": "推理过程", "plan": "计划", "criticism": "自我批评", "action": "行动名称", "action_input": "行动输入" }} """ response = self.llm.generate(prompt) try: return json.loads(response) except: return { "thought": "解析错误", "action": "finish", "action_input": "" } def execute(self, action: str, action_input: str) -> str: """执行行动""" if action == "finish": return "任务完成" if action in self.tools: result = self.tools[action](action_input) # 记录到记忆 self.memory.append({ "action": action, "input": action_input, "result": result }) return result else: return f"未知行动: {action}" def format_memory(self) -> str: """格式化记忆""" if not self.memory: return "No memories yet." return "\n".join([ f"- {m['action']}: {m['input']} -> {m['result'][:100]}" for m in self.memory[-5:] ]) def format_tasks(self) -> str: """格式化任务列表""" if not self.task_list: return "No tasks." return "\n".join([ f"{i+1}. {task}" for i, task in enumerate(self.task_list) ]) def run(self, max_iterations: int = 10) -> str: """运行Agent""" for i in range(max_iterations): # 思考 thought_process = self.think() print(f"\n=== Iteration {i+1} ===") print(f"Thought: {thought_process['thought']}") print(f"Reasoning: {thought_process['reasoning']}") print(f"Plan: {thought_process['plan']}") print(f"Criticism: {thought_process['criticism']}") # 执行 action = thought_process['action'] action_input = thought_process['action_input'] result = self.execute(action, action_input) print(f"Action: {action}") print(f"Result: {result[:200]}") # 检查是否完成 if action == "finish": return result return "达到最大迭代次数" # 使用示例 def search_web(query: str) -> str: """搜索网络""" return f"搜索'{query}'的结果..." def write_file(content: str) -> str: """写入文件""" return "文件已写入" def read_file(filename: str) -> str: """读取文件""" return f"文件{filename}的内容..." tools = { "search": search_web, "write": write_file, "read": read_file } agent = AutoGPTAgent( name="Researcher", role="AI研究员", goals=["研究最新AI技术", "生成研究报告"], llm=OpenAI(temperature=0), tools=tools ) result = agent.run() print(result) BabyAGI模式 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 class BabyAGI: """BabyAGI实现""" def __init__( self, objective: str, llm, tools: Dict[str, callable], max_iterations: int = 10 ): self.objective = objective self.llm = llm self.tools = tools self.max_iterations = max_iterations self.task_list = [] self.completed_tasks = [] def create_initial_tasks(self) -> List[str]: """创建初始任务列表""" prompt = f""" 目标: {self.objective} 请为这个目标创建一个任务列表。返回JSON数组格式： ["任务1", "任务2", "任务3"] """ response = self.llm.generate(prompt) try: tasks = json.loads(response) return tasks except: return ["研究相关资料", "分析问题", "制定方案"] def prioritize_tasks(self) -> List[str]: """任务优先级排序""" if not self.task_list: return [] prompt = f""" 目标: {self.objective} 当前任务列表: {json.dumps(self.task_list, ensure_ascii=False)} 已完成任务: {json.dumps(self.completed_tasks[-5:], ensure_ascii=False)} 请根据当前情况重新排列任务优先级。 返回JSON数组格式（从高到低）： ["任务1", "任务2", ...] """ response = self.llm.generate(prompt) try: return json.loads(response) except: return self.task_list def execute_task(self, task: str) -> str: """执行任务""" prompt = f""" 目标: {self.objective} 任务: {task} 请执行这个任务并返回结果。 如果需要使用工具，请说明： - search: 搜索信息 - calculate: 计算数据 - write: 写入内容 """ response = self.llm.generate(prompt) # 记录完成的任务 self.completed_tasks.append({ "task": task, "result": response }) return response def run(self) -> Dict: """运行BabyAGI""" # 创建初始任务 self.task_list = self.create_initial_tasks() for i in range(self.max_iterations): if not self.task_list: print("所有任务已完成！") break # 优先级排序 self.task_list = self.prioritize_tasks() # 执行第一个任务 current_task = self.task_list[0] print(f"\n=== 迭代 {i+1} ===") print(f"当前任务: {current_task}") result = self.execute_task(current_task) print(f"执行结果: {result[:200]}") # 从列表中移除 self.task_list.pop(0) return { "objective": self.objective, "completed_tasks": self.completed_tasks } # 使用 baby_agi = BabyAGI( objective="研究并总结2024年AI大模型的最新进展", llm=OpenAI(temperature=0), tools=tools ) result = baby_agi.run() CrewAI多Agent协作 Crew定义 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 from crewai import Agent, Task, Crew, Process # 定义Agent researcher = Agent( role='研究员', goal='研究最新的AI技术趋势', backstory="""你是一位经验丰富的AI研究员， 专注于追踪和分析最新的AI技术发展""", verbose=True, tools=[search_tool, wikipedia_tool] ) writer = Agent( role='技术作家', goal='将复杂的技术内容转化为易懂的文章', backstory="""你是一位技术写作专家， 擅长将技术细节转化为吸引人的内容""", verbose=True ) reviewer = Agent( role='内容审核员', goal='确保内容准确、完整、有价值', backstory="""你是一位资深的内容审核专家， 对技术内容的质量有极高要求""", verbose=True ) # 定义任务 research_task = Task( description="""研究2024年大语言模型的最新发展， 包括GPT-4、Claude、Gemini等模型的更新""", expected_output='详细的研究报告，包含关键发现和技术突破', agent=researcher ) write_task = Task( description="""基于研究报告，撰写一篇关于2024年LLM发展的技术文章。 文章应该面向技术读者，但保持通俗易懂""", expected_output='结构完整、内容丰富的技术文章（1000-1500字）', agent=writer ) review_task = Task( description="""审核技术文章，确保： 1. 技术准确性 2. 内容完整性 3. 可读性 4. 价值性 提供修改建议和最终评价""", expected_output='详细的审核报告，包含修改建议和最终评分', agent=reviewer ) # 创建Crew tech_crew = Crew( agents=[researcher, writer, reviewer], tasks=[research_task, write_task, review_task], process=Process.sequential, # 顺序执行 verbose=True ) # 执行 result = tech_crew.kickoff() print(result) 并行Process 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 # 定义并行任务 crew_parallel = Crew( agents=[agent1, agent2, agent3], tasks=[task1, task2, task3], process=Process.parallel, # 并行执行 verbose=True ) # 或者使用层级Process crew_hierarchical = Crew( agents=[manager_agent, worker_agent1, worker_agent2], tasks=[manager_task, worker_task1, worker_task2], process=Process.hierarchical, # 层级执行 manager_llm=OpenAI(temperature=0), verbose=True ) Agent编排框架对比 LangGraph 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 from langgraph.graph import StateGraph, END from typing import TypedDict # 定义状态 class AgentState(TypedDict): input: str research: str draft: str review: str final: str # 创建图 workflow = StateGraph(AgentState) # 添加节点 def research_node(state: AgentState) -> AgentState: result = researcher_agent.run(state["input"]) return {**state, "research": result} def write_node(state: AgentState) -> AgentState: result = writer_agent.run(state["research"]) return {**state, "draft": result} def review_node(state: AgentState) -> AgentState: result = reviewer_agent.run(state["draft"]) return {**state, "review": result} # 添加节点到图 workflow.add_node("researcher", research_node) workflow.add_node("writer", write_node) workflow.add_node("reviewer", review_node) # 添加边 workflow.set_entry_point("researcher") workflow.add_edge("researcher", "writer") workflow.add_edge("writer", "reviewer") workflow.add_edge("reviewer", END) # 编译图 app = workflow.compile() # 执行 result = app.invoke({"input": "研究AI最新进展"}) Semantic Kernel 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 from semantic_kernel import Kernel from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion from semantic_kernel.planning import SequentialPlanner # 初始化Kernel kernel = Kernel() kernel.add_chat_service( "chat-gpt", OpenAIChatCompletion("gpt-4", api_key="...") ) # 定义技能（Skill） from semantic_kernel.skill_definition import sk_function class ResearchSkills: @sk_function(description="搜索信息") def search(self, query: str) -> str: return f"搜索'{query}'的结果..." @sk_function(description="总结内容") def summarize(self, content: str) -> str: return f"总结: {content[:100]}..." # 注册技能 kernel.import_skill(ResearchSkills(), skill_name="research") # 创建计划器 planner = SequentialPlanner(kernel) # 执行计划 ask = "研究2024年AI技术进展并生成报告" plan = await planner.create_plan_async(ask) result = await plan.invoke_async(kernel) print(result) 实战案例 案例一：智能研究报告生成 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 class ReportGenerator: """智能报告生成器""" def __init__(self): self.llm = OpenAI(temperature=0) self.tools = self._init_tools() def _init_tools(self) -> Dict[str, callable]: return { "search": self._search, "analyze": self._analyze, "write": self._write, "format": self._format } def generate_report(self, topic: str) -> str: """生成报告""" # 阶段1：研究 research_data = self._research_stage(topic) # 阶段2：分析 analysis = self._analysis_stage(research_data) # 阶段3：撰写 draft = self._writing_stage(analysis) # 阶段4：审阅 final_report = self._review_stage(draft) return final_report def _research_stage(self, topic: str) -> Dict: """研究阶段""" # 生成研究计划 plan = self.llm.generate(f""" 为"{topic}"创建一个研究计划， 包含需要研究的关键点。 """) # 执行研究 research_data = {} for key_point in plan.split('\n'): if key_point.strip(): result = self._search(key_point.strip()) research_data[key_point.strip()] = result return research_data def _analysis_stage(self, data: Dict) -> str: """分析阶段""" prompt = f""" 分析以下研究数据： {json.dumps(data, ensure_ascii=False, indent=2)} 提供关键发现和洞察。 """ return self.llm.generate(prompt) def _writing_stage(self, analysis: str) -> str: """撰写阶段""" prompt = f""" 基于以下分析，撰写一份专业的研究报告： {analysis} 报告应该包含： 1. 执行摘要 2. 背景介绍 3. 主要发现 4. 结论和建议 """ return self.llm.generate(prompt) def _review_stage(self, draft: str) -> str: """审阅阶段""" prompt = f""" 审阅以下报告草稿： {draft} 提供改进建议并进行必要的修改。 """ reviewed = self.llm.generate(prompt) return reviewed def _search(self, query: str) -> str: """搜索实现""" # 实际调用搜索API return f"关于'{query}'的搜索结果..." 案例二：客户服务自动化 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 class CustomerServiceAgent: """客户服务Agent""" def __init__(self): self.llm = OpenAI(temperature=0.7) self.knowledge_base = self._load_kb() self.conversation_history = {} def handle_customer_query( self, customer_id: str, query: str ) -> str: """处理客户查询""" # 获取历史对话 history = self.conversation_history.get(customer_id, []) # 检索相关知识 relevant_docs = self._retrieve_knowledge(query) # 理解意图 intent = self._classify_intent(query) # 生成响应 response = self._generate_response( query=query, intent=intent, history=history, knowledge=relevant_docs ) # 更新历史 history.append({"role": "user", "content": query}) history.append({"role": "assistant", "content": response}) self.conversation_history[customer_id] = history[-10:] return response def _classify_intent(self, query: str) -> str: """分类意图""" prompt = f""" 分类以下客户查询的意图： 查询: {query} 可能的意图： 1. 产品咨询 2. 订单查询 3. 投诉建议 4. 售后服务 5. 其他 只返回意图名称。 """ return self.llm.generate(prompt).strip() def _retrieve_knowledge(self, query: str) -> List[str]: """检索相关知识""" # 使用向量搜索 # 简化示例 return [ doc for doc in self.knowledge_base if any(word in doc.lower() for word in query.lower().split()) ][:3] def _generate_response( self, query: str, intent: str, history: List[Dict], knowledge: List[str] ) -> str: """生成响应""" history_text = "\n".join([ f"{msg['role']}: {msg['content']}" for msg in history[-5:] ]) knowledge_text = "\n".join(knowledge) prompt = f""" 意图: {intent} 知识库: {knowledge_text} 对话历史: {history_text} 客户查询: {query} 请提供专业、友好的回复。 """ return self.llm.generate(prompt) 总结 AI Agent工作流编排是构建复杂AI应用的关键技术。从简单的LangChain Chains到复杂的Multi-Agent Systems，不同的框架和模式适用于不同的场景。 ...

引言 RAG（Retrieval-Augmented Generation，检索增强生成）技术是大模型应用的核心范式之一。它通过检索外部知识库来增强生成能力，有效解决了大模型知识滞后、幻觉严重等问题。本文将深入探讨RAG技术的完整实现链路，从文档处理到检索生成，帮助开发者构建企业级知识问答系统。 RAG技术概述 为什么需要RAG 1 2 3 4 5 6 7 8 9 10 11 12 # 纯LLM生成的问题 response = llm.generate("公司2024年Q3财报数据是多少？") # 问题1：模型知识截止，不知道最新信息 # 问题2：可能产生幻觉，编造数据 # RAG方案 relevant_docs = retriever.search("2024年Q3财报") response = llm.generate( prompt=f"基于以下文档回答问题：\n{relevant_docs}\n\n问题：公司2024年Q3财报数据是多少？" ) # 优势1：基于真实文档，准确可靠 # 优势2：可利用最新知识，实时更新 RAG核心流程 文档摄入 → 文本分块 → 向量化 → 向量数据库 ↓ 查询 → 向量化 → 检索 → 重排序 → LLM生成 → 答案 文档处理 1. 文档解析 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 from typing import List, Dict import pypdf from docx import Document import markdown class DocumentParser: """多格式文档解析器""" def parse(self, file_path: str) -> List[Dict]: """解析文档""" ext = self.get_extension(file_path) if ext == ".pdf": return self.parse_pdf(file_path) elif ext == ".docx": return self.parse_docx(file_path) elif ext in [".md", ".markdown"]: return self.parse_markdown(file_path) elif ext == ".txt": return self.parse_text(file_path) else: raise ValueError(f"Unsupported format: {ext}") def parse_pdf(self, file_path: str) -> List[Dict]: """解析PDF文件""" pages = [] with open(file_path, 'rb') as file: pdf_reader = pypdf.PdfReader(file) for page_num, page in enumerate(pdf_reader.pages): text = page.extract_text() # 提取元数据 metadata = { "page": page_num + 1, "source": file_path, "total_pages": len(pdf_reader.pages) } pages.append({ "content": text, "metadata": metadata }) return pages def parse_docx(self, file_path: str) -> List[Dict]: """解析Word文档""" doc = Document(file_path) paragraphs = [] for para_num, para in enumerate(doc.paragraphs): if para.text.strip(): paragraphs.append({ "content": para.text, "metadata": { "paragraph": para_num, "source": file_path } }) return paragraphs def parse_markdown(self, file_path: str) -> List[Dict]: """解析Markdown文件""" with open(file_path, 'r', encoding='utf-8') as f: md_content = f.read() # 使用markdown解析器 md = markdown.Markdown() html = md.convert(md_content) # 提取标题层级结构 sections = self.extract_sections(html, md_content) return sections def extract_sections(self, html: str, md: str) -> List[Dict]: """提取Markdown章节""" from bs4 import BeautifulSoup soup = BeautifulSoup(html, 'html.parser') sections = [] current_section = {"headers": [], "content": []} for element in soup.find_all(['h1', 'h2', 'h3', 'p', 'code']): if element.name.startswith('h'): # 保存之前的section if current_section["content"]: sections.append({ "content": "\n".join(current_section["content"]), "metadata": { "headers": current_section["headers"] } }) # 开始新的section level = int(element.name[1]) current_section = { "headers": current_section["headers"][:level-1] + [element.text], "content": [] } else: current_section["content"].append(element.get_text()) # 添加最后一个section if current_section["content"]: sections.append({ "content": "\n".join(current_section["content"]), "metadata": { "headers": current_section["headers"] } }) return sections 2. 文本分块 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 from typing import List import re from sentence_transformers import SentenceTransformer class TextChunker: """文本分块器""" def __init__(self, method: str = "recursive"): self.method = method self.embedding_model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2') def chunk(self, text: str, **kwargs) -> List[str]: """分块""" if self.method == "fixed_size": return self.fixed_size_chunk(text, **kwargs) elif self.method == "recursive": return self.recursive_chunk(text, **kwargs) elif self.method == "semantic": return self.semantic_chunk(text, **kwargs) else: raise ValueError(f"Unknown chunking method: {self.method}") def fixed_size_chunk( self, text: str, chunk_size: int = 500, overlap: int = 50 ) -> List[str]: """固定大小分块""" chunks = [] start = 0 text_length = len(text) while start < text_length: end = start + chunk_size chunk = text[start:end] chunks.append(chunk) # 移动到下一个块（考虑重叠） start = end - overlap return chunks def recursive_chunk( self, text: str, separators: List[str] = None, chunk_size: int = 1000, overlap: int = 100 ) -> List[str]: """递归分块 - 按分隔符智能分块""" if separators is None: separators = ["\n\n", "\n", "。", "！", "？", ".", "!", "?", " ", ""] # 尝试按分隔符分割 for separator in separators: if separator in text: parts = text.split(separator) chunks = [] current_chunk = "" for part in parts: # 如果添加这个部分会超过限制 if len(current_chunk) + len(part) + len(separator) > chunk_size: if current_chunk: chunks.append(current_chunk) # 如果单个部分本身就太长，递归处理 if len(part) > chunk_size: sub_chunks = self.recursive_chunk( part, separators[separators.index(separator) + 1:], chunk_size, overlap ) chunks.extend(sub_chunks) current_chunk = "" else: current_chunk = part + separator else: current_chunk += part + separator # 添加最后一个块 if current_chunk: chunks.append(current_chunk) return chunks # 如果没有找到分隔符，使用固定大小分块 return self.fixed_size_chunk(text, chunk_size, overlap) def semantic_chunk( self, text: str, max_similarity: float = 0.7 ) -> List[str]: """语义分块 - 基于语义相似度""" # 首先按句子分割 sentences = re.split(r'([。！？.!?])', text) sentences = [ s1 + s2 for s1, s2 in zip(sentences[::2], sentences[1::2]) if s1.strip() ] if not sentences: return [text] chunks = [] current_chunk = [sentences[0]] for sentence in sentences[1:]: # 计算与当前块最后一个句子的相似度 last_sentence = current_chunk[-1] similarity = self.compute_similarity(last_sentence, sentence) # 如果相似度高，合并到当前块 if similarity < max_similarity: current_chunk.append(sentence) else: # 否则开始新块 chunks.append("".join(current_chunk)) current_chunk = [sentence] # 添加最后一个块 if current_chunk: chunks.append("".join(current_chunk)) return chunks def compute_similarity(self, text1: str, text2: str) -> float: """计算语义相似度""" emb1 = self.embedding_model.encode(text1) emb2 = self.embedding_model.encode(text2) # 余弦相似度 import numpy as np return np.dot(emb1, emb2) / (np.linalg.norm(emb1) * np.linalg.norm(emb2)) 3. 元数据提取 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 class MetadataExtractor: """元数据提取器""" def __init__(self): self.date_patterns = [ r'\d{4}-\d{2}-\d{2}', r'\d{4}年\d{1,2}月\d{1,2}日', r'\d{4}/\d{2}/\d{2}' ] self.url_pattern = r'https?://[^\s]+' self.email_pattern = r'\w+@\w+\.\w+' def extract(self, chunk: str) -> Dict: """提取元数据""" metadata = {} # 提取日期 metadata['dates'] = self.extract_dates(chunk) # 提取URL metadata['urls'] = self.extract_urls(chunk) # 提取邮箱 metadata['emails'] = self.extract_emails(chunk) # 提取关键词 metadata['keywords'] = self.extract_keywords(chunk) # 提取实体 metadata['entities'] = self.extract_entities(chunk) return metadata def extract_dates(self, text: str) -> List[str]: """提取日期""" dates = [] for pattern in self.date_patterns: dates.extend(re.findall(pattern, text)) return dates def extract_urls(self, text: str) -> List[str]: """提取URL""" return re.findall(self.url_pattern, text) def extract_emails(self, text: str) -> List[str]: """提取邮箱""" return re.findall(self.email_pattern, text) def extract_keywords(self, text: str, top_k: int = 5) -> List[str]: """提取关键词""" # 使用TF-IDF或RAKE算法 from sklearn.feature_extraction.text import TfidfVectorizer vectorizer = TfidfVectorizer(max_features=top_k) tfidf_matrix = vectorizer.fit_transform([text]) feature_names = vectorizer.get_feature_names_out() tfidf_scores = tfidf_matrix.toarray()[0] # 获取top-k关键词 top_indices = tfidf_scores.argsort()[-top_k:][::-1] return [feature_names[i] for i in top_indices] def extract_entities(self, text: str) -> Dict[str, List[str]]: """提取命名实体""" # 使用spaCy或其他NER工具 import spacy nlp = spacy.load("zh_core_web_sm") doc = nlp(text) entities = { "PERSON": [], "ORG": [], "GPE": [], "DATE": [] } for ent in doc.ents: if ent.label_ in entities: entities[ent.label_].append(ent.text) return entities 向量数据库 1. ChromaDB集成 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 import chromadb from chromadb.config import Settings class ChromaVectorStore: """ChromaDB向量存储""" def __init__(self, collection_name: str = "documents"): self.client = chromadb.Client(Settings( chroma_db_impl="duckdb+parquet", persist_directory="./chroma_db" )) # 创建或获取collection self.collection = self.client.get_or_create_collection( name=collection_name, metadata={"hnsw:space": "cosine"} ) def add_documents( self, documents: List[str], embeddings: List[List[float]], metadatas: List[Dict], ids: List[str] ): """添加文档""" self.collection.add( documents=documents, embeddings=embeddings, metadatas=metadatas, ids=ids ) def search( self, query_embedding: List[float], top_k: int = 5, where: Dict = None ) -> Dict: """搜索相似文档""" results = self.collection.query( query_embeddings=[query_embedding], n_results=top_k, where=where ) return { "documents": results["documents"][0], "metadatas": results["metadatas"][0], "distances": results["distances"][0] } def delete(self, ids: List[str]): """删除文档""" self.collection.delete(ids=ids) def update( self, ids: List[str], documents: List[str] = None, embeddings: List[List[float]] = None, metadatas: List[Dict] = None ): """更新文档""" self.collection.update( ids=ids, documents=documents, embeddings=embeddings, metadatas=metadatas ) 2. Pinecone集成 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 import pinecone from pinecone import ServerlessSpec class PineconeVectorStore: """Pinecone向量存储""" def __init__(self, api_key: str, environment: str): pinecone.init(api_key=api_key, environment=environment) self.index_name = "rag-index" self.dimension = 768 # 根据embedding模型调整 # 创建index如果不存在 if self.index_name not in pinecone.list_indexes(): pinecone.create_index( name=self.index_name, dimension=self.dimension, metric="cosine", spec=ServerlessSpec( cloud="aws", region="us-east-1" ) ) self.index = pinecone.Index(self.index_name) def upsert(self, vectors: List[Dict]): """批量插入/更新向量""" self.index.upsert(vectors=vectors) def query( self, vector: List[float], top_k: int = 5, filter: Dict = None, include_metadata: bool = True ) -> Dict: """查询相似向量""" return self.index.query( vector=vector, top_k=top_k, filter=filter, include_metadata=include_metadata ) def delete(self, ids: List[str]): """删除向量""" self.index.delete(ids=ids) 3. Milvus集成 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType class MilvusVectorStore: """Milvus向量存储""" def __init__(self, host: str = "localhost", port: int = 19530): # 连接Milvus connections.connect("default", host=host, port=port) # 定义collection schema self.collection_name = "rag_documents" self.dimension = 768 fields = [ FieldSchema(name="id", dtype=DataType.VARCHAR, is_primary=True, max_length=100), FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=self.dimension), FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=65535), FieldSchema(name="metadata", dtype=DataType.JSON) ] schema = CollectionSchema(fields, f"{self.collection_name} schema") # 创建collection if self.collection_name not in [c.name for c in connections.list_connections()]: self.collection = Collection( name=self.collection_name, schema=schema ) else: self.collection = Collection(self.collection_name) # 创建索引 index_params = { "index_type": "IVF_FLAT", "metric_type": "COSINE", "params": {"nlist": 128} } self.collection.create_index( field_name="embedding", index_params=index_params ) self.collection.load() def insert(self, data: List[Dict]): """插入数据""" self.collection.insert(data) def search( self, embedding: List[float], top_k: int = 5, expr: str = None ) -> Dict: """搜索""" results = self.collection.search( data=[embedding], anns_field="embedding", param={"metric_type": "COSINE", "params": {"nprobe": 10}}, limit=top_k, expr=expr, output_fields=["content", "metadata"] ) return results[0] 检索策略 1. 语义检索 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 from sentence_transformers import SentenceTransformer import numpy as np class SemanticRetriever: """语义检索器""" def __init__( self, vector_store: ChromaVectorStore, model_name: str = "paraphrase-multilingual-MiniLM-L12-v2" ): self.vector_store = vector_store self.embedding_model = SentenceTransformer(model_name) def retrieve( self, query: str, top_k: int = 5, filters: Dict = None ) -> List[Dict]: """检索相关文档""" # 生成查询向量 query_embedding = self.embedding_model.encode(query) # 向量搜索 results = self.vector_store.search( query_embedding=query_embedding.tolist(), top_k=top_k, where=filters ) # 格式化结果 documents = [] for i, doc in enumerate(results["documents"]): documents.append({ "content": doc, "metadata": results["metadatas"][i], "score": 1 - results["distances"][i] # 转换为相似度 }) return documents 2. 混合检索 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 class HybridRetriever: """混合检索器（语义+关键词）""" def __init__( self, vector_store: ChromaVectorStore, keyword_index: object, # Elasticsearch或Whoosh semantic_weight: float = 0.7 ): self.vector_store = vector_store self.keyword_index = keyword_index self.semantic_weight = semantic_weight def retrieve( self, query: str, top_k: int = 5 ) -> List[Dict]: """混合检索""" # 语义检索 semantic_results = self.semantic_search(query, top_k * 2) # 关键词检索 keyword_results = self.keyword_search(query, top_k * 2) # 归一化分数 semantic_scores = self.normalize_scores([r["score"] for r in semantic_results]) keyword_scores = self.normalize_scores([r["score"] for r in keyword_results]) # 合并分数 combined_scores = {} for doc, score in zip(semantic_results, semantic_scores): doc_id = doc["metadata"]["id"] combined_scores[doc_id] = score * self.semantic_weight for doc, score in zip(keyword_results, keyword_scores): doc_id = doc["metadata"]["id"] if doc_id in combined_scores: combined_scores[doc_id] += score * (1 - self.semantic_weight) else: combined_scores[doc_id] = score * (1 - self.semantic_weight) # 排序并返回top-k sorted_docs = sorted( combined_scores.items(), key=lambda x: x[1], reverse=True )[:top_k] return [self.get_document(doc_id) for doc_id, _ in sorted_docs] def normalize_scores(self, scores: List[float]) -> List[float]: """归一化分数到0-1""" min_score = min(scores) max_score = max(scores) if max_score == min_score: return [1.0] * len(scores) return [ (score - min_score) / (max_score - min_score) for score in scores ] 3. 重排序 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 from sentence_transformers import CrossEncoder class Reranker: """重排序器""" def __init__(self, model_name: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"): self.reranker = CrossEncoder(model_name) def rerank( self, query: str, documents: List[Dict], top_k: int = 5 ) -> List[Dict]: """重新排序""" # 准备query-document对 pairs = [ (query, doc["content"]) for doc in documents ] # 计算重排序分数 scores = self.reranker.predict(pairs) # 添加分数到文档 for doc, score in zip(documents, scores): doc["rerank_score"] = float(score) # 按重排序分数排序 documents.sort(key=lambda x: x["rerank_score"], reverse=True) return documents[:top_k] 生成增强 1. Prompt构建 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 class PromptBuilder: """Prompt构建器""" def __init__(self, template_type: str = "qa"): self.template_type = template_type def build( self, query: str, context: List[Dict], chat_history: List[Dict] = None ) -> str: """构建prompt""" if self.template_type == "qa": return self.build_qa_prompt(query, context) elif self.template_type == "conversational": return self.build_conversational_prompt(query, context, chat_history) elif self.template_type == "structured": return self.build_structured_prompt(query, context) else: raise ValueError(f"Unknown template type: {self.template_type}") def build_qa_prompt(self, query: str, context: List[Dict]) -> str: """构建QA prompt""" # 格式化上下文 context_text = "\n\n".join([ f"文档{i+1}:\n{doc['content']}\n来源: {doc['metadata'].get('source', 'unknown')}" for i, doc in enumerate(context) ]) prompt = f""" 请基于以下文档内容回答问题。如果文档中没有相关信息，请明确说明。 ### 参考文档 {context_text} ### 问题 {query} ### 回答要求 1. 只使用参考文档中的信息 2. 如果文档中没有答案，明确说明"根据提供的文档，我无法回答这个问题" 3. 引用具体来源 4. 保持准确和客观 回答： """ return prompt.strip() def build_conversational_prompt( self, query: str, context: List[Dict], chat_history: List[Dict] ) -> str: """构建对话式prompt""" context_text = "\n\n".join([ f"- {doc['content']}" for doc in context ]) # 格式化历史对话 history_text = "\n".join([ f"{msg['role']}: {msg['content']}" for msg in chat_history[-5:] # 只保留最近5轮 ]) prompt = f""" ### 参考信息 {context_text} ### 对话历史 {history_text} ### 当前问题 {query} ### 要求 1. 基于参考信息回答 2. 考虑对话历史上下文 3. 保持自然对话风格 4. 如果参考信息不足，可以基于常识补充 回答： """ return prompt.strip() def build_structured_prompt(self, query: str, context: List[Dict]) -> str: """构建结构化prompt（用于生成结构化输出）""" context_text = "\n\n".join([ f"【{doc['metadata'].get('title', '文档')}】\n{doc['content']}" for doc in context ]) prompt = f""" ## 任务说明 请基于以下参考文档回答用户问题，并按指定格式输出。 ## 参考文档 {context_text} ## 用户问题 {query} ## 输出格式要求 请按以下JSON格式输出： {{ "answer": "详细回答", "confidence": "高/中/低", "sources": ["来源1", "来源2"], "key_points": ["要点1", "要点2"] }} 请输出： """ return prompt.strip() 2. LLM生成 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 import openai from typing import Dict, List class RAGGenerator: """RAG生成器""" def __init__( self, api_key: str, model: str = "gpt-4", temperature: float = 0.7 ): openai.api_key = api_key self.model = model self.temperature = temperature def generate( self, query: str, context: List[Dict], stream: bool = False ) -> Dict: """生成回答""" # 构建prompt prompt_builder = PromptBuilder(template_type="qa") prompt = prompt_builder.build(query, context) # 调用LLM response = openai.ChatCompletion.create( model=self.model, messages=[ { "role": "system", "content": "你是一个专业的知识助手，擅长基于提供的文档回答问题。" }, { "role": "user", "content": prompt } ], temperature=self.temperature, stream=stream ) if stream: # 流式输出 return self._stream_response(response) else: # 一次性返回 answer = response.choices[0].message.content return { "answer": answer, "sources": [doc["metadata"] for doc in context], "model": self.model } def _stream_response(self, response): """处理流式响应""" for chunk in response: if chunk.choices[0].delta.get("content"): yield chunk.choices[0].delta.content 完整RAG系统 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 class RAGSystem: """完整RAG系统""" def __init__(self, config: Dict): # 初始化组件 self.embedder = SentenceTransformer(config["embedding_model"]) self.vector_store = ChromaVectorStore(config["collection_name"]) self.retriever = SemanticRetriever(self.vector_store) self.reranker = Reranker(config["reranker_model"]) self.generator = RAGGenerator(config["openai_api_key"]) self.prompt_builder = PromptBuilder(template_type="qa") # 文档处理 self.parser = DocumentParser() self.chunker = TextChunker(method="recursive") self.metadata_extractor = MetadataExtractor() def ingest_documents(self, file_paths: List[str]): """摄取文档""" all_chunks = [] for file_path in file_paths: # 解析文档 documents = self.parser.parse(file_path) # 分块 for doc in documents: chunks = self.chunker.chunk( doc["content"], chunk_size=1000, overlap=200 ) # 提取元数据 for i, chunk in enumerate(chunks): metadata = { **doc["metadata"], "chunk_index": i, "source_file": file_path } # 额外元数据提取 extra_metadata = self.metadata_extractor.extract(chunk) metadata.update(extra_metadata) all_chunks.append({ "content": chunk, "metadata": metadata }) # 生成嵌入 embeddings = self.embedder.encode([c["content"] for c in all_chunks]) # 存储到向量数据库 ids = [f"doc_{i}" for i in range(len(all_chunks))] self.vector_store.add_documents( documents=[c["content"] for c in all_chunks], embeddings=embeddings.tolist(), metadatas=[c["metadata"] for c in all_chunks], ids=ids ) return len(all_chunks) def query( self, question: str, top_k: int = 5, rerank: bool = True ) -> Dict: """查询""" # 检索 retrieved_docs = self.retriever.retrieve( query=question, top_k=top_k * 2 # 检索更多用于重排序 ) # 重排序 if rerank: retrieved_docs = self.reranker.rerank( query=question, documents=retrieved_docs, top_k=top_k ) # 生成回答 response = self.generator.generate( query=question, context=retrieved_docs ) return response def chat( self, message: str, chat_history: List[Dict] = None, top_k: int = 3 ) -> Dict: """对话模式""" # 从历史中提取上下文 if chat_history: # 可以使用历史对话优化检索 context_query = self._build_context_query(message, chat_history) else: context_query = message # 检索 retrieved_docs = self.retriever.retrieve( query=context_query, top_k=top_k ) # 构建对话式prompt prompt_builder = PromptBuilder(template_type="conversational") prompt = prompt_builder.build(message, retrieved_docs, chat_history) # 生成 response = self.generator.generate( query=message, context=retrieved_docs ) # 添加到历史 if chat_history is None: chat_history = [] chat_history.append({"role": "user", "content": message}) chat_history.append({"role": "assistant", "content": response["answer"]}) response["chat_history"] = chat_history return response def _build_context_query(self, message: str, history: List[Dict]) -> str: """基于历史构建上下文查询""" # 提取历史中的关键词 recent_messages = history[-4:] # 最近2轮对话 context = " ".join([ msg["content"] for msg in recent_messages if msg["role"] == "user" ]) return f"{context} {message}" 优化策略 1. 查询扩展 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 class QueryExpander: """查询扩展器""" def __init__(self, llm): self.llm = llm def expand(self, query: str, num_expansions: int = 3) -> List[str]: """扩展查询""" prompt = f""" 请为以下查询生成{num_expansions}个语义相似但表述不同的查询版本。 原始查询: {query} 请只输出扩展后的查询，每行一个： """ response = self.llm.generate(prompt) expanded_queries = [query] expanded_queries.extend([ q.strip() for q in response.split('\n') if q.strip() ][:num_expansions]) return expanded_queries # 使用 expanded_queries = query_expander.expand("如何提高RAG系统性能？") # ["如何提高RAG系统性能？", "优化RAG检索效果的方法", "RAG系统性能提升技巧"] 2. Hybrid Search 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 class HybridSearchRetriever: """混合检索（稠密+稀疏）""" def retrieve(self, query: str, top_k: int = 10) -> List[Dict]: # 稠密检索（向量） dense_results = self.dense_retriever.retrieve(query, top_k * 2) # 稀疏检索（BM25） sparse_results = self.sparse_retriever.retrieve(query, top_k * 2) # RRF（Reciprocal Rank Fusion）合并 fused_results = self.rrf_fusion( dense_results, sparse_results, k=60 ) return fused_results[:top_k] def rrf_fusion( self, results1: List[Dict], results2: List[Dict], k: int = 60 ) -> List[Dict]: """RRF融合算法""" scores = {} # 计算第一个结果的分数 for rank, doc in enumerate(results1): doc_id = doc["id"] scores[doc_id] = 1 / (k + rank + 1) # 叠加第二个结果的分数 for rank, doc in enumerate(results2): doc_id = doc["id"] if doc_id in scores: scores[doc_id] += 1 / (k + rank + 1) else: scores[doc_id] = 1 / (k + rank + 1) # 排序 sorted_docs = sorted(scores.items(), key=lambda x: x[1], reverse=True) return [self.get_doc(doc_id) for doc_id, _ in sorted_docs] 3. 缓存机制 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 from functools import lru_cache import hashlib class CachedRAGSystem(RAGSystem): """带缓存的RAG系统""" @lru_cache(maxsize=1000) def _cached_retrieve(self, query_hash: str, top_k: int): """缓存检索结果""" return super().retrieve(query_hash, top_k) def query(self, question: str, top_k: int = 5) -> Dict: # 生成查询hash query_hash = hashlib.md5(question.encode()).hexdigest() # 尝试从缓存获取 try: cached_result = self._cached_retrieve(query_hash, top_k) return cached_result except: # 缓存未命中，执行正常检索 result = super().query(question, top_k) return result 评估指标 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 class RAGEvaluator: """RAG系统评估器""" def __init__(self, test_data: List[Dict]): self.test_data = test_data def evaluate(self, rag_system: RAGSystem) -> Dict: """评估RAG系统""" metrics = { "retrieval_precision": [], "retrieval_recall": [], "answer_relevance": [], "faithfulness": [] } for test_case in self.test_data: question = test_case["question"] ground_truth_docs = test_case["relevant_docs"] ground_truth_answer = test_case["answer"] # 检索评估 retrieved_docs = rag_system.retriever.retrieve(question, top_k=10) retrieved_doc_ids = [doc["metadata"]["id"] for doc in retrieved_docs] precision = self.compute_precision( retrieved_doc_ids, ground_truth_docs ) recall = self.compute_recall( retrieved_doc_ids, ground_truth_docs ) metrics["retrieval_precision"].append(precision) metrics["retrieval_recall"].append(recall) # 生成评估 response = rag_system.query(question) relevance = self.compute_relevance( response["answer"], ground_truth_answer ) faithfulness = self.compute_faithfulness( response["answer"], retrieved_docs ) metrics["answer_relevance"].append(relevance) metrics["faithfulness"].append(faithfulness) # 计算平均指标 return { "avg_precision": np.mean(metrics["retrieval_precision"]), "avg_recall": np.mean(metrics["retrieval_recall"]), "avg_relevance": np.mean(metrics["answer_relevance"]), "avg_faithfulness": np.mean(metrics["faithfulness"]) } def compute_precision(self, retrieved: List, relevant: List) -> float: """计算精确率""" retrieved_set = set(retrieved) relevant_set = set(relevant) if not retrieved_set: return 0.0 return len(retrieved_set & relevant_set) / len(retrieved_set) def compute_recall(self, retrieved: List, relevant: List) -> float: """计算召回率""" retrieved_set = set(retrieved) relevant_set = set(relevant) if not relevant_set: return 0.0 return len(retrieved_set & relevant_set) / len(relevant_set) def compute_relevance(self, generated: str, reference: str) -> float: """计算答案相关性（使用余弦相似度）""" from sentence_transformers import SentenceTransformer model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2') emb1 = model.encode(generated) emb2 = model.encode(reference) import numpy as np return np.dot(emb1, emb2) / (np.linalg.norm(emb1) * np.linalg.norm(emb2)) def compute_faithfulness(self, answer: str, contexts: List[Dict]) -> float: """计算忠实度（答案是否基于检索的上下文）""" # 使用LLM判断 prompt = f""" 请判断以下答案是否完全基于提供的上下文信息。 上下文: {chr(10).join([c['content'] for c in contexts[:3]])} 答案: {answer} 请回答"是"或"否"，并简要说明理由。 """ response = self.llm.generate(prompt) return 1.0 if "是" in response else 0.0 总结 RAG技术通过结合检索和生成，为大模型应用提供了强大的知识增强能力。本文详细介绍了从文档处理、向量存储、检索策略到生成增强的完整技术链路。 ...

引言 AI Agent（人工智能智能体）作为大语言模型最重要的应用范式之一，正在重塑我们与AI交互的方式。不同于传统的聊天机器人，AI Agent具备自主感知、决策和执行能力，能够使用工具、维护记忆、进行多步推理。本文将深入探讨AI Agent的架构设计，从理论到实践，帮助开发者构建生产级的智能体应用。 AI Agent核心概念 什么是AI Agent AI Agent是一个能够： 感知环境：理解用户输入和系统状态 推理决策：基于目标和上下文制定行动方案 执行工具：调用外部API和服务完成任务 记忆管理：维护短期和长期记忆 反思学习：从执行结果中学习和改进 Agent vs Chatbot 1 2 3 4 5 6 7 8 9 10 11 12 # 传统Chatbot chatbot_response = llm.generate("帮我查询天气") # 单轮对话，无状态，无法执行操作 # AI Agent agent = Agent( tools=[weather_api, calendar_api], memory=LongTermMemory(), planner=ReActPlanner() ) result = agent.run("帮我查明天天气，如果有雨则安排线上会议") # 多步推理，工具调用，状态管理 核心架构设计 1. 整体架构 ┌─────────────────────────────────────────────┐ │ User Interface │ └──────────────────┬──────────────────────────┘ │ ┌──────────────────▼──────────────────────────┐ │ Agent Core │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ Planner │──│ Executor │──│ Reflector│ │ │ └──────────┘ └──────┬───┘ └──────────┘ │ │ │ │ │ ┌─────────────────────┼─────────────────┐ │ │ │ Memory System │ │ │ │ ┌─────────┐ ┌─────────┐ ┌────────┐ │ │ │ │ │ShortTerm│ │LongTerm │ │Vector │ │ │ │ │ └─────────┘ └─────────┘ └────────┘ │ │ │ └────────────────────────────────────────┘ │ └──────────────────┬──────────────────────────┘ │ ┌──────────────────▼──────────────────────────┐ │ Tool Layer │ │ ┌─────────┐ ┌─────────┐ ┌─────────────┐ │ │ │ API │ │ Database│ │ Functions │ │ │ │ Calls │ │ Query │ │ Execution │ │ │ └─────────┘ └─────────┘ └─────────────┘ │ └─────────────────────────────────────────────┘ 2. Planner模块 规划器负责将用户目标分解为可执行的步骤。 ...