多 Collection 隔离 混合索引 动态 CRUD架构最近在技术社区看到一个常见问题为什么我的 AI 助手明明安装了向量数据库但无法记住历史对话原因在于采用RAG 架构的AI助手最初只是为 LLM 提供外部知识查询能力不是为动态记忆管理设计的。那么拥有动态记忆的AI助手要如何搭建RAG架构、agentic RAG、agent的演化路线是如何的我们要如何对其进行选型以及记忆能力的配置本文将梳理从 RAG 到智能体记忆的演进路径以及 Milvus 在每个阶段的核心使用方式。01第一阶段RAG——只读的外部知识库RAG 的本质是什么RAG检索增强生成本质就是给 LLM 配备外部知识库让它回答问题前先查询资料。2020 年 Lewis 等人提出这个概念用于解决 LLM 的知识截止日期问题。RAG 的工作流程先在离线阶段将文档切分成小块并转换为向量存入数据库运行时将用户问题也转换为向量并检索最相似的 Top-K 文档最后将检索结果和用户问题一起输入 LLM 生成答案。不难发现在实现 RAG 时最大的技术挑战是如何在百万级/亿级向量中实现毫秒级检索传统数据库在这个场景下存在明显短板要么不支持向量索引要么只采用简单的向量检索插件在百万级数据上检索延迟巨大无法满足实时性要求。Milvus在内向量数据库则可以 提供专业的向量索引HNSW、IVF_FLAT 等做到百亿数据的毫秒级延迟。Milvus 实现 RAG 的核心代码如下from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType import openai # 工具函数统一的向量化接口 def embed(text): 将文本转换为向量 response openai.Embedding.create( inputtext, modeltext-embedding-ada-002 ) return response[data][0][embedding] # 步骤1连接Milvus并创建Collection connections.connect(hostlocalhost, port19530) fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(nametext, dtypeDataType.VARCHAR, max_length65535), FieldSchema(nameembedding, dtypeDataType.FLOAT_VECTOR, dim1536), FieldSchema(namesource, dtypeDataType.VARCHAR, max_length512), ] schema CollectionSchema(fieldsfields, descriptionRAG Knowledge Base) collection Collection(namerag_knowledge, schemaschema) # 步骤2创建索引 index_params { index_type: HNSW, metric_type: COSINE, params: {M: 16, efConstruction: 256} } collection.create_index(field_nameembedding, index_paramsindex_params) # 步骤3离线索引——批量插入文档 def ingest_documents(documents): 批量插入文档到向量库 texts [] embeddings [] sources [] for doc in documents: texts.append(doc[text]) embeddings.append(embed(doc[text])) sources.append(doc[source]) data [texts, embeddings, sources] collection.insert(data) collection.flush() # 确保数据持久化 print(f✅ 已索引 {len(documents)} 条文档) # 步骤4在线检索——RAG查询 def rag_retrieval(query, top_k5): 传统RAG检索每次必检索 collection.load() # 加载到内存 query_embedding embed(query) search_params {metric_type: COSINE, params: {ef: 100}} results collection.search( data[query_embedding], anns_fieldembedding, paramsearch_params, limittop_k, output_fields[text, source] ) return results[0] # 使用示例 docs [ {text: Milvus是开源向量数据库支持HNSW索引, source: docs/intro.md}, {text: RAG通过检索增强生成提升LLM能力, source: docs/rag.md} ] ingest_documents(docs) results rag_retrieval(什么是向量数据库) for hit in results: print(f相似度: {hit.score:.3f} | 内容: {hit.entity.get(text)}) 通过以上架构示意图和代码我们可以发现在传统的simple RAG架构中每次查询都强制检索并且知识库在离线阶段构建后运行时无法更新且所有知识来自同一向量库无法动态切换数据源。这些问题的根源在于RAG 把检索当作必选项而不是可选的精细化分类的工具。这些问题在简单问答场景影响不大但在复杂智能体系统中会成为瓶颈。那么能否让系统像人类一样根据问题类型决定是否需要查资料这就是agentic RAG 的核心思想。02第二阶段agentic RAG —按需检索传统simple RAG 每次查询都强制调用检索不管是否真的需要额外知识。agentic RAG的突破是把检索变成可选工具Agent 自主决策是否需要检索、检索什么来源、结果是否可信。相应的在这个过程中我们需要引入多 Collection 架构。因为单 Collection 架构存在明显问题检索结果混杂不同领域内容不同领域的向量混在同一语义空间导致召回准确率下降且无法针对不同类型知识设置差异化检索策略。多 Collection 方案的优势在于每个领域独立索引product_docs、api_reference、customer_cases等Agent 根据问题类型精准路由不同领域使用同的索引参数和过滤条件。下面通过实际代码展示如何用 Milvus 实现这套多 Collection 架构。核心要点是为每个领域创建独立 CollectionAgent 根据问题类型动态路由检索。from pymilvus import connections, Collection connections.connect(hostlocalhost, port19530) # 创建多个专业领域的Collection class MultiSourceRAG: def __init__(self): self.collections { product_docs: Collection(product_docs), # 产品文档 api_reference: Collection(api_reference), # API参考 customer_cases: Collection(customer_cases), # 客户案例 tech_blogs: Collection(tech_blogs) # 技术博客 } # 加载所有Collection到内存 for coll in self.collections.values(): coll.load() # Agent决策智能检索路由 def smart_retrieve(question, agent_decision): Agent决策示例 { need_retrieval: True, target_collections: [api_reference, tech_blogs], top_k: 5, filters: {publish_date: 2024-01-01} } if not agent_decision[need_retrieval]: return [] # Agent判断不需要检索 rag MultiSourceRAG() results [] for coll_name in agent_decision[target_collections]: collection rag.collections[coll_name] # 构建动态过滤表达式 filter_expr None if filters in agent_decision: filters agent_decision[filters] if publish_date in filters: filter_expr fpublish_date {filters[publish_date]} # 执行检索 search_params {metric_type: IP, params: {nprobe: 16}} search_results collection.search( data[embed(question)], anns_fieldembedding, paramsearch_params, limitagent_decision[top_k], exprfilter_expr, # Milvus支持标量过滤 output_fields[text, source, publish_date] ) results.extend(search_results[0]) return results # 检索质量评估 def retrieve_with_quality_check(question, threshold0.7): Agent评估检索质量决定下一步行动 collection Collection(product_docs) collection.load() results collection.search( data[embed(question)], anns_fieldembedding, param{metric_type: IP, params: {nprobe: 16}}, limit5 ) # 过滤低质量结果 high_quality_results [ hit for hit in results[0] if hit.score threshold ] # Agent决策 if not high_quality_results: return {action: FALLBACK_TO_WEB_SEARCH, reason: 本地知识库召回质量不足} return {action: USE_RESULTS, data: high_quality_results}尽管agentic RAG 在检索决策上实现了突破但仍有一个核心问题未解决智能体 RAG 的核心问题未解决知识库依然是只读的。Agent 可以决定什么时候读、读什么但不能写入新知识、更新旧知识、删除过时知识。这引出了下一阶段agent memory系统。03第三阶段agent memoryAgent 记忆需要完整的 CRUD 能力实时保存对话中的偏好和事件检索历史会话中的相关记忆修正用户提供的新信息清理过期或无效记录。这要求底层存储系统支持运行时的写入和更新操作。但是实践中不同类型的记忆无法使用统一策略。比如用户说我喜欢简洁的回复是长期偏好需保留数月甚至数年但今天天气怎么样这类对话只需保留几天。如果混合存储会导致查询用户沟通偏好时结果混杂大量无关对话检索精度下降无法设置差异化过期策略要么误删长期偏好要么历史对话无限膨胀拖垮性能。解决方案是按生命周期分类程序性记忆 Collection 存储长期偏好importance 0.8情景记忆 Collection 存储对话历史30-90 天过期语义记忆 Collection 存储事实知识长期有效可修正。以下是借助Milvus 实现多 Collection 隔离 混合索引 动态 CRUD架构如何应用于agent memory的参考。from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection from datetime import datetime connections.connect(hostlocalhost, port19530) # Collection Schema定义 def create_memory_collection(name, description): 创建标准化的记忆Collection fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(nameuser_id, dtypeDataType.VARCHAR, max_length64), FieldSchema(namecontent, dtypeDataType.VARCHAR, max_length5000), FieldSchema(nameembedding, dtypeDataType.FLOAT_VECTOR, dim768), FieldSchema(nameimportance, dtypeDataType.FLOAT), FieldSchema(namecreated_at, dtypeDataType.INT64), FieldSchema(namemetadata, dtypeDataType.JSON), ] schema CollectionSchema(fieldsfields, descriptiondescription) collection Collection(namename, schemaschema) # 创建向量索引 collection.create_index( field_nameembedding, index_params{index_type: HNSW, metric_type: IP, params: {M: 16}} ) # 创建标量索引加速user_id过滤 collection.create_index(field_nameuser_id, index_params{index_type: TRIE}) collection.load() return collection # 初始化三种记忆类型 class AgentMemorySystem: def __init__(self): self.procedural_memory create_memory_collection( procedural_memory, 用户偏好与行为规则 ) self.episodic_memory create_memory_collection( episodic_memory, 对话历史与事件记录 ) self.semantic_memory create_memory_collection( semantic_memory, 事实性知识 ) memory_system AgentMemorySystem()核心操作记忆写入Createdef store_memory(memory_type, user_id, content, importance0.5, metadataNone): 实时写入记忆 # 选择对应的Collection if memory_type procedural: collection memory_system.procedural_memory elif memory_type episodic: collection memory_system.episodic_memory else: collection memory_system.semantic_memory # 准备数据 data [{ user_id: user_id, content: content, embedding: embed(content), importance: importance, created_at: int(datetime.now().timestamp()), metadata: metadata or {} }] # 实时插入 collection.insert(data) collection.flush() # 确保持久化 print(f✅ 已存储{memory_type}记忆: {content[:50]}...) # 使用场景 # 场景1Agent从对话中提取用户偏好 store_memory( memory_typeprocedural, user_iduser_123, content用户喜欢简洁的回复多用emoji, importance0.8, metadata{category: communication_style} ) # 场景2记录对话事件 store_memory( memory_typeepisodic, user_iduser_123, content用户提到10月30日要去巴黎旅行需要推荐景点, importance0.9, metadata{event_type: travel_plan, date: 2024-10-30} ) # 场景3存储事实性知识 store_memory( memory_typesemantic, user_iduser_123, content埃菲尔铁塔位于法国巴黎高330米建于1889年, importance0.7, metadata{entity: 埃菲尔铁塔, source: wikipedia} )核心操作记忆检索Readdef retrieve_memories(user_id, query, memory_typeall, top_k5, min_importance0.3): 智能记忆检索支持多类型过滤 query_embedding embed(query) results {} # 构建过滤表达式用户隔离 重要性过滤 filter_expr fuser_id {user_id} importance {min_importance} search_params {metric_type: IP, params: {ef: 100}} # 选择要检索的Collection collections_to_search [] if memory_type in [all, procedural]: collections_to_search.append((procedural, memory_system.procedural_memory)) if memory_type in [all, episodic]: collections_to_search.append((episodic, memory_system.episodic_memory)) if memory_type in [all, semantic]: collections_to_search.append((semantic, memory_system.semantic_memory)) # 执行检索 for mem_type, collection in collections_to_search: search_results collection.search( data[query_embedding], anns_fieldembedding, paramsearch_params, limittop_k, exprfilter_expr, output_fields[content, importance, created_at, metadata] ) results[mem_type] search_results[0] return results # 使用场景 memories retrieve_memories( user_iduser_123, query用户想了解巴黎旅游信息, memory_typeall, min_importance0.7 ) print( 召回的记忆:) for mem_type, hits in memories.items(): print(f\n【{mem_type}】:) for hit in hits[:2]: print(f 相似度: {hit.score:.3f} | {hit.entity.get(content)[:60]}...)核心操作记忆更新与删除Update Deletedef update_memory(collection, memory_id, new_content, new_importanceNone): 更新记忆Milvus 2.3支持upsert 注意生产环境需考虑一致性问题 当前先删后插方案存在风险 - 删除成功但插入失败 → 记忆丢失 - 并发更新 → 数据竞争 推荐使用Milvus的upsert操作原子性 # 先删除旧记忆 collection.delete(exprfid {memory_id}) # 插入新记忆 data [{ content: new_content, embedding: embed(new_content), importance: new_importance or 0.5, created_at: int(datetime.now().timestamp()) }] collection.insert(data) collection.flush() def forget_memory(collection, criteria): 选择性遗忘记忆 策略示例 - 时间衰减删除90天前的低重要性情景记忆 - 置信度过滤删除置信度0.6的语义记忆 # 示例删除过期的情景记忆 if older_than_days in criteria: cutoff_time int(datetime.now().timestamp()) - criteria[older_than_days] * 86400 filter_expr fcreated_at {cutoff_time} importance {criteria.get(min_importance, 0.5)} collection.delete(exprfilter_expr) print(f️ 已清理 {criteria[older_than_days]} 天前的低重要性记忆) # 使用场景 # 场景1用户修正信息 update_memory( collectionmemory_system.episodic_memory, memory_id12345, new_content用户旅行时间改为11月15日, new_importance0.9 ) # 场景2定期清理过期记忆 forget_memory( collectionmemory_system.episodic_memory, criteria{older_than_days: 90, min_importance: 0.4} )04三个阶段的本质差异以下是从 RAG 到记忆的技术演进总结回顾整个演进过程核心变化体现在两个维度数据更新时机和操作权限。传统 RAG 的知识库在离线阶段完成索引构建运行时只支持查询操作。如果知识内容需要更新必须停止服务重新构建索引。agentic RAG 引入了检索决策机制。系统会先判断是否需要检索、以及从哪个数据源检索避免了无效查询。但数据本身依然是只读的运行时无法修改知识库内容。Agent memory 阶段实现了关键突破系统具备了运行时的写入能力。它可以在对话过程中创建新记忆、更新过期信息、删除无用记录。这种从只读到读写的转变使得系统能够动态维护用户的个性化知识库。回顾这三个阶段的演进有个反直觉的发现技术难度在下降工程难度在上升。传统 RAG 的核心是怎么检索得更准——索引算法、相似度计算。到了智能体记忆阶段技术问题反而简单了Milvus 把 CRUD 都封装好了真正棘手的是工程决策什么信息值得记记多久什么时候更新什么时候忘掉实际项目中很多团队没想清楚记忆策略。用户的随口抱怨要不要记记错了怎么办三个月不登录的用户记忆要不要清理这些问题没有标准答案完全取决于业务场景。真正的壁垒是对用户记忆管理的理解深度。最后为什么要学AI大模型当下⼈⼯智能市场迎来了爆发期并逐渐进⼊以⼈⼯通⽤智能AGI为主导的新时代。企业纷纷官宣“ AI ”战略为新兴技术⼈才创造丰富的就业机会⼈才缺⼝将达 400 万DeepSeek问世以来生成式AI和大模型技术爆发式增长让很多岗位重新成了炙手可热的新星岗位薪资远超很多后端岗位在程序员中稳居前列。与此同时AI与各行各业深度融合飞速发展成为炙手可热的新风口企业非常需要了解AI、懂AI、会用AI的员工纷纷开出高薪招聘AI大模型相关岗位。最近很多程序员朋友都已经学习或者准备学习 AI 大模型后台也经常会有小伙伴咨询学习路线和学习资料我特别拜托北京清华大学学士和美国加州理工学院博士学位的鲁为民老师给大家这里给大家准备了一份涵盖了AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频全系列的学习资料这些学习资料不仅深入浅出而且非常实用让大家系统而高效地掌握AI大模型的各个知识点。这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】AI大模型系统学习路线在面对AI大模型开发领域的复杂与深入精准学习显得尤为重要。一份系统的技术路线图不仅能够帮助开发者清晰地了解从入门到精通所需掌握的知识点还能提供一条高效、有序的学习路径。但知道是一回事做又是另一回事初学者最常遇到的问题主要是理论知识缺乏、资源和工具的限制、模型理解和调试的复杂性在这基础上找到高质量的学习资源不浪费时间、不走弯路又是重中之重。AI大模型入门到实战的视频教程项目包看视频学习是一种高效、直观、灵活且富有吸引力的学习方式可以更直观地展示过程能有效提升学习兴趣和理解力是现在获取知识的重要途径光学理论是没用的要学会跟着一起敲要动手实操才能将自己的所学运用到实际当中去这时候可以搞点实战案例来学习。海量AI大模型必读的经典书籍PDF阅读AI大模型经典书籍可以帮助读者提高技术水平开拓视野掌握核心技术提高解决问题的能力同时也可以借鉴他人的经验。对于想要深入学习AI大模型开发的读者来说阅读经典书籍是非常有必要的。600AI大模型报告实时更新这套包含640份报告的合集涵盖了AI大模型的理论研究、技术实现、行业应用等多个方面。无论您是科研人员、工程师还是对AI大模型感兴趣的爱好者这套报告合集都将为您提供宝贵的信息和启示。AI大模型面试真题答案解析我们学习AI大模型必然是想找到高薪的工作下面这些面试题都是总结当前最新、最热、最高频的面试题并且每道题都有详细的答案面试前刷完这套面试题资料小小offer不在话下这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】