LLM

深入探讨AI辅助开发工具的技术原理和最佳实践，包括代码生成、智能调试、重构建议等功能

深入探讨AI Agent的设计原理与开发实践，帮助你构建能够自主规划、调用工具、协同工作的智能代理系统

引言 AI技术正在革命性地改变游戏内容的创作方式。从程序化生成到大型语言模型驱动的动态内容，AI正在帮助开发者创造更丰富、更个性化的游戏体验。本文将深入探讨AI在游戏内容生成中的各种应用。 程序化生成基础 PCG技术 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 """ 程序化内容生成 (PCG) 随机数生成: - 伪随机: 可重现 - 真随机: 真实随机性 - 噪声函数: Perlin, Simplex 生成方法: - 基于规则: 语法, L-System - 基于搜索: 遗传算法, 模拟退火 - 基于学习: 神经网络, GAN """ class PCGFoundation: """程序化生成基础""" def __init__(self): self.techniques = { "随机数": { "伪随机": "种子可重现", "噪声函数": "Perlin, Simplex, Worley", "应用": "地形, 纹理生成" }, "规则系统": { "语法": "字符串重写", "L-System": "分形生成", "波浪函数坍缩": "约束满足" }, "学习生成": { "GAN": "生成对抗网络", "VAE": "变分自编码器", "Diffusion": "扩散模型" } } def terrain_generation(self): """地形生成""" methods = { "Perlin噪声": { "原理": "梯度噪声叠加", "优势": "自然连续", "应用": "高度图生成" }, "元胞自动机": { "原理": "局部规则演化", "应用": "洞穴生成", "优势": "简单有效" }, "Voronoi图": { "原理": "区域划分", "应用": "生物群系", "优势": "自然分区" } } return methods AI关卡生成 深度学习应用 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 class AILevelGeneration: """AI关卡生成""" def __init__(self): self.applications = { "2D平台": { "GAN": "训练于真实关卡", "VAE": "潜在空间探索", "RL": "强化学习生成" }, "3D环境": { "神经网络": "从图像生成3D", "风格迁移": "艺术风格应用", "NeRF": "神经辐射场" } } def dungeon_generation(self): """地牢生成""" methods = { "传统": { "算法": "随机游走, 元胞自动机", "优势": "快速可控", "限制": "模式有限" }, "AI增强": { "GAN": "学习地牢模式", "RL": "优化可玩性", "混合": "传统+AI" }, "评估指标": { "可玩性": "可达路径", "趣味性": "挑战分布", "美学": "视觉平衡" } } return methods 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 class LLMStoryGeneration: """LLM剧情生成""" def __init__(self): self.components = { "故事引擎": { "情节生成": "主线故事", "对话生成": "角色对话", "任务生成": "支线任务" }, "世界模拟": { "NPC": "独立行为", "事件": "动态事件", "因果": "因果关系" } } def narrative_generation(self): """叙事生成""" generation = { "故事结构": { "英雄之旅": "经典叙事", "分支": "玩家选择影响", "涌现": "系统产生故事" }, "LLM应用": { "情节": "生成故事大纲", "对话": "角色对话", "描述": "场景描述" }, "一致性": { "记忆": "角色记忆", "状态": "世界状态", "约束": "逻辑约束" } } return generation def quest_generation_example(self): """任务生成示例""" example = { "输入": { "玩家等级": 15, "位置": "暗影森林", "背景": "附近有强盗出没" }, "LLM生成": { "任务名": "森林的威胁", "描述": "村民受强盗骚扰", "目标": ["击败5个强盗", "找到营地", "击败头目"], "奖励": ["金币100", "经验500", "短剑"] }, "动态": "基于世界状态生成" } return example 动态难度调整 AI驱动的平衡 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 class DynamicDifficulty: """动态难度""" def __init__(self): self.methods = { "玩家建模": { "技能": "评估玩家水平", "行为": "分析游戏模式", "预测": "预测玩家行为" }, "难度调整": { "敌人": "强度, 数量", "奖励": "资源掉落", "环境": "可用的帮助" }, "AI学习": { "强化学习": "优化难度", "玩家反馈": "满意度学习", "个性化": "个人化难度" } } def player_profiling(self): """玩家画像""" profiling = { "技能水平": { "新手": "引导, 简化", "中级": "平衡", "专家": "挑战, 隐藏内容" }, "游戏风格": { "探索": "奖励探索", "战斗": "更多战斗", "社交": "社交互动" }, "实时调整": { "死亡": "降低难度", "轻松": "提高难度", "流畅": "保持当前" } } return profiling AI辅助开发 开发工具 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 class AIAssistedDev: """AI辅助开发""" def __init__(self): self.tools = { "代码生成": { "脚本": "行为树, AI脚本", "配置": "游戏平衡参数", "工具": "Copilot, ChatGPT" }, "资产生成": { "2D": "精灵图, 纹理", "3D": "模型, 动画", "音频": "音效, 音乐" }, "测试": { "自动化": "AI测试玩家", "平衡": "数值平衡测试", "Bug": "异常检测" } } def asset_generation_tools(self): """资产生成工具""" tools = { "图像生成": { "工具": ["Midjourney", "Stable Diffusion", "DALL-E"], "应用": ["概念图", "纹理", "UI元素"] }, "3D生成": { "工具": ["Shap-E", "Point-E", "TripoSR"], "应用": ["道具", "环境", "角色"] }, "音频生成": { "工具": ["MusicLM", "AudioLDM"], "应用": ["背景音乐", "音效", "语音"] } } return tools 实际应用案例 成功案例 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 class CaseStudies: """应用案例""" def __init__(self): self.cases = { "No Man's Sky": { "技术": "程序化生成", "规模": "18万亿星球", "AI": "算法生成生态" }, "AI Dungeon": { "技术": "GPT驱动", "类型": "文字冒险", "特色": "无限可能" }, "Courtship": { "技术": "AI社交", "特色": "智能NPC", "应用": "社交模拟" } } def implementation_lessons(self): """实施经验""" lessons = { "混合方法": { "优势": "AI+传统", "平衡": "可控+创造力", "建议": "不要完全依赖AI" }, "玩家测试": { "重要性": "验证质量", "反馈": "改进AI", "迭代": "持续优化" }, "性能": { "考虑": "实时生成", "缓存": "预生成", "优化": "性能影响" } } return lessons 挑战与限制 技术挑战 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 class AIContentChallenges: """AI内容生成挑战""" def __init__(self): self.challenges = { "质量控制": { "问题": "AI生成质量不稳定", "解决": "人工审核, 自动评估", "平衡": "创意vs质量" }, "一致性": { "问题": "前后不一致", "解决": "上下文约束", "技术": "记忆机制" }, "性能": { "问题": "实时生成延迟", "解决": "预生成, 缓存", "优化": "模型优化" } } def future_solutions(self): """未来解决方案""" solutions = { "多模态": { "技术": "视觉+语言+音频", "应用": "完整体验" }, "个性化": { "技术": "玩家模型", "应用": "定制内容" }, "共创": { "模式": "AI+人工", "应用": "增强创作" } } return solutions 未来展望 发展趋势 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 class AIContentFuture: """AI内容生成未来""" def __init__(self): self.trends = { "更高质量": { "模型": "更大更强", "精度": "细节提升", "一致性": "更好的上下文" }, "实时生成": { "技术": "边缘计算", "优化": "模型压缩", "应用": "无限游戏" }, "玩家共创": { "模式": "玩家提示AI", "平台": "UGC平台", "生态": "创作社区" } } def emerging_applications(self): """新兴应用""" applications = { "虚拟世界": { "技术": "AI生成元宇宙", "规模": "无限内容", "演进": "持续变化" }, "个性化游戏": { "技术": "AI适应玩家", "体验": "独特体验", "留存": "长期参与" }, "协作创作": { "技术": "AI辅助创作", "工作流": "开发者+AI", "效率": "10x提升" } } return applications 总结 AI正在革命性地改变游戏内容的创作方式。从程序化生成到大型语言模型，AI技术为游戏带来了无限的可能性。虽然还存在质量和一致性等挑战，但随着技术的进步，AI将在游戏开发中发挥越来越重要的作用。 ...

引言 大语言模型的突破为游戏NPC带来了革命性的变化。传统的脚本化NPC正在被具备自然对话能力、记忆和情感的智能角色所取代。本文将深入探讨如何将LLM集成到游戏中，设计具有个性和记忆的NPC角色，并构建高效稳定的对话系统。 LLM NPC基础架构 系统设计概述 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 """ LLM驱动的NPC系统架构 核心组件: 1. LLM引擎: 对话生成 2. 记忆系统: 长期和短期记忆 3. 人格模块: 角色设定 4. 上下文管理: 对话历史 5. 游戏集成: 与游戏世界交互 """ class LLMMNPCSystem: """LLM NPC系统""" def __init__(self): self.architecture = { "LLM引擎": { "模型选择": "GPT-4, Claude, 或开源模型", "部署方式": "API调用或本地部署", "推理优化": "量化, 蒸馏, 缓存" }, "人格系统": { "基础设定": "背景, 性格, 目标", "对话风格": "口吻, 习惯用语", "知识库": "角色相关知识" }, "记忆系统": { "短期记忆": "当前对话", "长期记忆": "向量数据库", "记忆检索": "语义搜索" }, "游戏集成": { "事件触发": "游戏事件映射", "状态同步": "NPC状态更新", "动作执行": "对话转动作" } } def prompt_engineering(self, character_name): """提示工程""" prompt_template = """ 你是一个游戏中的NPC角色，名字叫{name}。 角色背景： {background} 性格特点： {personality} 你的说话风格： {style} 当前场景： {scene} 玩家刚才说： {player_input} 请以角色的身份回应玩家，保持角色一致性。 记住： 1. 保持角色性格 2. 回复要简洁（1-3句话） 3. 如果玩家询问游戏相关信息，可以适度透露 4. 如果玩家提出不合理要求，用角色的方式拒绝 """ return prompt_template def build_npc_context(self, npc_id, conversation_history): """构建NPC上下文""" context = { "系统提示": self.get_system_prompt(npc_id), "角色信息": self.get_character_info(npc_id), "当前状态": self.get_npc_state(npc_id), "位置": self.get_npc_location(npc_id), "对话历史": conversation_history[-10:], # 保留最近10轮 "记忆检索": self.retrieve_relevant_memories(npc_id) } return context 角色人格设计 角色塑造系统 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 class CharacterPersonality: """角色人格系统""" def __init__(self): self.personality_dimensions = { "大五人格": { "开放性": "好奇心, 创造力", "尽责性": "纪律, 可靠性", "外向性": "社交, 热情", "宜人性": "合作, 同理心", "神经质": "情绪稳定性" }, "对话风格": { "正式度": "正式 vs 随意", "幽默感": "严肃 vs 幽默", "话多": "简短 vs 健谈", "方言": "标准口音 vs 方言" } } def create_character_profile(self, character_data): """创建角色档案""" profile = { "基础信息": { "名字": character_data["name"], "年龄": character_data["age"], "职业": character_data["profession"], "背景": character_data["backstory"] }, "性格特征": { "人格维度": character_data["personality"], "价值观": character_data["values"], "动机": character_data["motivations"], "恐惧": character_data["fears"] }, "对话特征": { "说话方式": character_data["speech_style"], "口头禅": character_data["catchphrases"], "话题偏好": character_data["topics"], "禁忌话题": character_data["taboo_topics"] }, "知识领域": { "专业技能": character_data["skills"], "世界观": character_data["world_knowledge"], "人际关系": character_data["relationships"] } } return profile def personality_example(self): """角色人格示例""" merchant_npc = { "名字": "老汤姆", "职业": "商人", "性格": { "开放性": "高 - 见多识广", "尽责性": "中 - 时而精明时而马虎", "外向性": "高 - 喜欢聊天", "宜人性": "中 - 看人说话", "神经质": "低 - 乐观" }, "说话风格": { "正式度": "随意", "口头禅": ["哈哈！", "好买卖！", "年轻人"], "特征": "热情但精明" }, "系统提示": """ 你是老汤姆，一个在幻想世界经营了30年的商人。 你见过无数冒险者，知道这个世界的各种秘密。 你的说话风格热情随意，经常说"哈哈！"和"好买卖！"。 虽然你看起来友好，但本质上是个精明的商人，不会做亏本生意。 你会用各种方式推销商品，但不会强买强卖。 """ } return merchant_npc 记忆系统设计 长期和短期记忆 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 class NPCMemorySystem: """NPC记忆系统""" def __init__(self): self.memory_types = { "短期记忆": { "存储": "当前会话", "容量": "最近10-20轮对话", "保留": "会话结束可清除" }, "长期记忆": { "存储": "向量数据库", "类型": ["重要对话", "玩家行为", "游戏事件"], "检索": "语义搜索" }, "情景记忆": { "内容": "特定事件和经历", "重要性": "情感强度相关", "衰减": "时间衰减" }, "语义记忆": { "内容": "世界知识和常识", "稳定性": "不随时间衰减", "共享": "NPC间可共享" } } def memory_embedding(self, text): """记忆嵌入""" # 使用嵌入模型将文本转向量 embedding = { "模型": "text-embedding-ada-002", "维度": 1536, "用途": "语义相似度计算" } return embedding def store_memory(self, npc_id, memory_data, importance=0.5): """存储记忆""" memory = { "npc_id": npc_id, "content": memory_data["content"], "type": memory_data["type"], # conversation, event, observation "timestamp": time.time(), "importance": importance, "emotion": memory_data.get("emotion", "neutral"), "entities": memory_data.get("entities", []), "embedding": self.memory_embedding(memory_data["content"]) } # 存储到向量数据库 self.vector_store.add(memory) return memory def retrieve_memories(self, npc_id, query, top_k=5): """检索相关记忆""" # 1. 将查询转向量 query_embedding = self.memory_embedding(query) # 2. 向量搜索 similar_memories = self.vector_store.search( npc_id=npc_id, embedding=query_embedding, top_k=top_k ) # 3. 考虑时间衰减 current_time = time.time() for memory in similar_memories: age = current_time - memory["timestamp"] decay = math.exp(-age / (30 * 24 * 3600)) # 30天衰减 memory["retrieval_score"] *= decay return similar_memories def memory_consolidation(self, npc_id): """记忆巩固""" # 定期将短期记忆转为长期记忆 short_term = self.get_short_term_memories(npc_id) for memory in short_term: # 评估重要性 importance = self.evaluate_importance(memory) if importance > 0.7: # 转为长期记忆 self.store_memory(npc_id, memory, importance) 对话系统实现 对话管理器 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 class NPCDialogueManager: """NPC对话管理器""" def __init__(self, llm_client, memory_system): self.llm = llm_client self.memory = memory_system self.conversation_sessions = {} def start_conversation(self, npc_id, player_id): """开始对话""" session_id = f"{npc_id}_{player_id}_{int(time.time())}" self.conversation_sessions[session_id] = { "npc_id": npc_id, "player_id": player_id, "messages": [], "start_time": time.time(), "state": "active" } # 获取NPC上下文 context = self.build_npc_context(npc_id) return session_id, context def process_input(self, session_id, player_input): """处理玩家输入""" session = self.conversation_sessions[session_id] # 1. 检索相关记忆 relevant_memories = self.memory.retrieve_memories( session["npc_id"], player_input ) # 2. 构建提示 prompt = self.build_prompt( session["npc_id"], player_input, session["messages"], relevant_memories ) # 3. LLM生成 try: response = self.llm.generate( prompt=prompt, max_tokens=150, temperature=0.8, stop_sequences=["\n", "玩家:"] ) npc_response = { "text": response, "timestamp": time.time(), "memories_accessed": len(relevant_memories) } # 4. 更新对话历史 session["messages"].append({ "role": "player", "content": player_input }) session["messages"].append({ "role": "npc", "content": response }) # 5. 存储记忆 self.memory.store_memory( session["npc_id"], { "content": f"玩家说: {player_input}\n我回应: {response}", "type": "conversation", "emotion": self.detect_emotion(response) }, importance=self.calculate_importance(player_input) ) return npc_response except Exception as e: # 降级处理 return self.fallback_response(session["npc_id"]) def build_prompt(self, npc_id, player_input, history, memories): """构建完整提示""" prompt = f""" {self.get_system_prompt(npc_id)} 相关记忆： {self.format_memories(memories)} 对话历史： {self.format_history(history)} 玩家说：{player_input} 你的回应： """ return prompt def detect_emotion(self, text): """检测对话情感""" # 简单情感检测 emotions = { "开心": ["高兴", "哈哈", "太好了", "喜欢"], "生气": ["气死", "讨厌", "滚", "烦"], "悲伤": ["难过", "伤心", "可惜"], "惊讶": ["什么？", "天哪", "真的"] } detected = "neutral" for emotion, keywords in emotions.items(): if any(kw in text for kw in keywords): detected = emotion break return detected 游戏集成 与游戏世界交互 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 class GameWorldIntegration: """游戏世界集成""" def __init__(self, dialogue_manager): self.dialogue = dialogue_manager self.action_executor = ActionExecutor() def handle_dialogue_action(self, session_id, npc_response): """处理对话触发的动作""" # 1. 解析NPC意图 intent = self.parse_intent(npc_response["text"]) # 2. 执行相应动作 if intent["type"] == "trade": self.action_executor.open_trade(session_id) elif intent["type"] == "quest": quest_id = intent.get("quest_id") self.action_executor.offer_quest(session_id, quest_id) elif intent["type"] == "give_item": item_id = intent.get("item_id") self.action_executor.give_item(session_id, item_id) elif intent["type"] == "attack": self.action_executor.attack_player(session_id) def trigger_npc_behavior(self, npc_id, trigger_event): """触发NPC行为""" npc_state = self.get_npc_state(npc_id) # 根据事件和NPC状态生成行为 prompt = f""" 你是{npc_id}，当前状态：{npc_state} 发生的事件：{trigger_event} 你会如何反应？请描述你的行动和说话。 """ response = self.dialogue.llm.generate(prompt) # 执行生成的行为 self.execute_npc_action(npc_id, response) def dynamic_quest_generation(self, npc_id, player_context): """动态生成任务""" prompt = f""" 你是{npc_id}，一个NPC。 玩家的情况：{player_context} 根据你的性格和知识，为玩家设计一个合适的任务。 任务应包括： 1. 任务名称 2. 任务描述 3. 任务目标 4. 任务奖励 请以JSON格式返回。 """ quest_data = self.dialogue.llm.generate(prompt) # 解析并创建任务 quest = self.parse_quest(quest_data) return quest 性能优化 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 63 64 65 66 67 68 69 70 71 72 class LLMOptimization: """LLM优化策略""" def __init__(self): self.strategies = { "缓存": { "响应缓存": "相似问题复用", "嵌入缓存": "减少重复计算", "实现": "Redis缓存" }, "批处理": { "批量请求": "合并多个请求", "流水线": "异步处理", "实现": "消息队列" }, "模型优化": { "量化": "INT8量化", "蒸馏": "小模型", "剪枝": "移除冗余" }, "本地部署": { "优势": "低延迟", "模型": "开源7B-13B模型", "硬件": "GPU推理" } } def response_cache_strategy(self): """响应缓存策略""" cache = { "相似度匹配": { "方法": "余弦相似度", "阈值": 0.85, "命中": 直接返回缓存 }, "缓存结构": { "key": "对话特征向量", "value": "NPC响应", "ttl": "24小时" }, "更新策略": { "LRU": "最近最少使用", "LFU": "最不经常使用", "TTL": "时间过期" } } return cache def cost_optimization(self): """成本优化""" optimization = { "模型选择": { "简单对话": "3B-7B模型", "复杂推理": "13B-30B模型", "创意任务": "GPT-4/Claude" }, "提示优化": { "精简提示": "减少token", "系统提示缓存": "不重复发送", "few-shot": "精选示例" }, "请求合并": { "批量处理": "多个请求一次", "异步": "非阻塞处理", "优先级": "重要请求优先" } } return optimization 实际应用案例 RPG游戏中的智能NPC 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 class RPGNPCImplementation: """RPG游戏NPC实现""" def __init__(self): self.example_scenario = { "场景": "奇幻酒馆", "NPC": "酒馆老板玛拉", "功能": [ "提供信息和传言", "发布任务", "买卖物品", "推进剧情" ] } def dialogue_example(self): """对话示例""" example = """ 玩家: 听说最近附近有盗贼出没？ 玛拉: [检索记忆: 玩家询问当地治安] [检索记忆: 最近商队被劫] 是的，年轻人。商队的损失不小，[压低声音]我听说领主正在招募勇敢的人去调查这事。 你对这个感兴趣吗？如果是的话，或许我可以帮你引见。 玩家: 我很感兴趣，但需要什么装备？ 玛拉: [识别玩家需求: 准备任务] 哈哈！我就知道你是个有胆量的。你需要一些基本装备... [触发: 打开商店界面] 看看我这里的武器，虽然不是最好的，但对付几个盗贼足够了。 [情感: 热情，精明] """ return example def quest_generation_example(self): """任务生成示例""" quest = { "任务名称": "商队的复仇", "给予者": "玛拉", "描述": "最近商队频频遭袭，领主悬赏调查盗贼巢穴", "目标": [ "前往盗贼出没的森林", "找到盗贼营地", "消灭盗贼头目或带回情报" ], "奖励": { "金币": 100, "经验": 500, "物品": "玛拉的感谢信（可换折扣）" }, "对话生成": "基于玛拉的性格和当前情况动态生成" } return quest 挑战与解决方案 常见问题 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 class LLMMPCChallenges: """LLM NPC挑战与解决""" def __init__(self): self.challenges = { "一致性问题": { "问题": "NPC性格前后不一致", "解决": [ "强化系统提示", "长期记忆引用", "人格约束", "后处理验证" ] }, "延迟问题": { "问题": "LLM响应慢", "解决": [ "本地部署小模型", "流式输出", "预测性预生成", "降级方案" ] }, "成本问题": { "问题": "API调用成本高", "解决": [ "响应缓存", "模型分层", "本地部署", "批处理" ] }, "安全性": { "问题": "生成不当内容", "解决": [ "内容过滤", "输出审查", "角色约束", "人工审核" ] } } def fallback_strategies(self): """降级策略""" fallbacks = { "LLM失败": "使用预设对话树", "响应超时": "返回通用回应", "成本过高": "切换到小模型", "不当内容": "安全默认回复" } return fallbacks 未来展望 技术发展趋势 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 class LLMMPCFuture: """LLM NPC未来展望""" def __init__(self): self.trends = { "多模态": { "技术": "视觉+语音+文本", "应用": "面部表情，动作", "体验": "更真实的交互" }, "持续学习": { "技术": "从玩家交互学习", "个性化": "适应不同玩家", "进化": "角色动态成长" }, "社交智能": { "技术": "NPC间社交网络", "群体": "群体行为涌现", "动态": "动态关系变化" }, "情感计算": { "技术": "情感建模", "表达": "情感驱动对话", "深度": "更深的情感连接" } } def emerging_applications(self): """新兴应用""" applications = { "虚拟主播": { "应用": "游戏中的虚拟主播", "技术": "LLM+TTS+面部驱动", "交互": "实时观众互动" }, "动态剧情": { "应用": "根据玩家选择生成剧情", "技术": "LLM剧情生成", "体验": "个性化故事" }, "智能副本": { "应用": "AI驱动的副本设计", "技术": "程序化生成+LLM", "重玩": "无限重玩价值" } } return applications 总结 大语言模型为游戏NPC带来了前所未有的智能化可能。从简单的对话树到具有记忆、情感和个性的智能角色，LLM NPC正在重新定义玩家与游戏世界的交互方式。随着技术成熟和成本下降，我们将会看到更多游戏采用这一技术，创造更丰富、更沉浸的游戏体验。 ...

引言 多模态大模型（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），大幅降低了微调成本和门槛。 ...

引言 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模块 规划器负责将用户目标分解为可执行的步骤。 ...

引言 2025年，大语言模型已经彻底改变了软件开发的方式。我们正在见证从"手写代码"到"人机协作"的范式转移。这不是简单的工具升级，而是软件工程方法论的根本性变革。本文将深入分析这一变革，帮助开发者适应新时代的开发模式。 一、软件工程的范式转移 1.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 # 软件工程1.0：开发者即编码者 class TraditionalDeveloper: """ 角色定位： - 80%时间写代码 - 15%时间调试 - 5%时间设计 核心技能： - 语法熟练度 - API记忆 - 手动调试 - 文档查阅 """ def create_feature(self, requirement): # 1. 手写数据模型 class User: def __init__(self, id, name, email): self.id = id self.name = name self.email = email # 2. 手写数据访问层 class UserRepository: def get_by_id(self, id): # 手写SQL pass # 3. 手写API端点 def user_endpoint(request): # 手写路由处理 pass # 4. 手写测试 def test_user(): # 手写测试用例 pass # ...所有代码都需要手动编写 AI辅助开发模式 ...

深入探讨AI Agent的架构设计模式，涵盖ReAct、CoT、多智能体协作等核心模式，帮助开发者构建更强大的AI应用系统。