嘉峪关市网站建设_网站建设公司_外包开发_seo优化

Agent 的状态数据分两种：会话内的临时上下文和跨会话的长期知识。

前者是“用户刚才说了什么”、“工具返回了什么结果”，会话结束就该清空。后者是“用户三个月前的购买记录”、“历史工单的解决方案”，需要持久化并在未来对话中智能召回。

混用这两种数据会导致很多用户体验上的问题，比如：把临时上下文永久化，数据库会堆满无用的工具调用记录。把长期知识当临时数据处理，用户下次登录时 Agent 会忘记所有偏好。

而要解决这个问题，需要在架构层面将存储和检索逻辑做好分离。

在这一背景下，Google 在 2025 年开源了 Agent Development Kit（ADK），这是一个生产级的 Agent 开发框架：能够在架构层面强制分离会话内上下文和跨会话知识，避免开发者把这两种数据混在一起。

那么其运行逻辑是什么？如何使用，本文将重点解答。

01

ADK的设计逻辑：状态管理一等公民化

ADK的核心设计逻辑在于，把状态管理作为框架的核心设计。

每个 Agent 实例天生带着SessionService（会话内临时数据）和MemoryService（跨会话长期知识）两个接口，避免了状态对象在代码层级间传递。

其中SessionService 支持内存、数据库、云托管三种后端，MemoryService 则是接口规范，具体存储后端由开发者选择。

整体架构一共分三层：Agent 层定义智能体类型，Runtime 层管理 Session 生命周期和 Event 流，Service 层抽象外部依赖。核心是 Event 驱动模型，Agent 的每次操作被封装成 Event 追加到事件流，可以在任意时刻回放决策过程。

ADK 通过state的 prefix 机制细化数据作用域：user:前缀在用户的所有 Session 间共享，app:前缀在应用级别共享，temp:前缀仅在当前 Session 有效。

02

Milvus：ADK 的记忆层实现

ADK 的 MemoryService 需要三个重点能力：语义检索、混合查询、亿级扩展性。

Milvus 在这三个维度上分别支持：

语义检索上，Milvus 支持 HNSW 和 DiskANN等十多种针对不同场景的高效向量索引算法，并能做到千万级数据下查询延迟可以控制在 10ms 以内。

混合检索上，Milvus 支持向量搜索+标量过滤；语义+关键词等针对不同内容、不同数据类型的高效检索，且不影响整体性能与召回率。

扩展性方面，Milvus 的计算存储分离架构支持 Query Node 的动态扩展，单机可以支撑千万级向量。

本文 Demo 主要使用 Milvus Lite或Milvus Standalone 进行本地开发测试。

03

实战：从 0 到 1 搭建带 Milvus 记忆的 Agent

实现一个技术客服 Agent，需求是：当用户提问时，Agent 能从历史工单中召回相似案例，避免重复解答。这个场景之所以典型，是因为它涉及了 Agent 记忆系统的三大难点：跨会话检索（用户今天问的问题，可能和上个月的工单相关）、用户隔离（不能把用户 A 的记忆召回给用户 B)、语义匹配（用户的提问方式千变万化，不能用精确匹配）。

环境准备：

1. Python3.11+

2.docker、dockercompos

3. Gemini-API-KEY

这部分是基础配置，确保程序能正常运行

pip install google-adk pymilvus google-generativeai

“”“ ADK + Milvus + Gemini 长期记忆 Agent 演示如何实现跨会话的记忆召回系统 ”“” import os import asyncio import time from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType, utility import google.generativeai as genai from google.adk.agents import Agent from google.adk.tools import FunctionTool from google.adk.runners import Runner from google.adk.sessions import InMemorySessionService from google.genai import types

1.部署 Milvus Standalone（Docker）

1.1下载部署文件

wget <https://github.com/Milvus-io/Milvus/releases/download/v2.5.12/Milvus-standalone-docker-compose.yml> -O docker-compose.yml 1.2启动Milvus服务

docker-compose up -d

docker-compose ps -a

2.模型和连接配置

配置 Gemini API 和 Milvus 连接参数

# ==================== 配置部分 ==================== # 1. Gemini API 配置 GOOGLE_API_KEY = os.getenv(“GOOGLE_API_KEY”) if not GOOGLE_API_KEY: raise ValueError(“请设置环境变量 GOOGLE_API_KEY”) genai.configure(api_key=GOOGLE_API_KEY) # 2. Milvus 连接配置 MILVUS_HOST = os.getenv(“MILVUS_HOST”, “localhost”) MILVUS_PORT = os.getenv(“MILVUS_PORT”, “19530”) # 3. 模型选择（免费层级额度最高的组合） LLM_MODEL = “gemini-2.5-flash-lite” # LLM 模型：1000RPD EMBEDDING_MODEL = “models/text-embedding-004” # Embedding 模型：1000 RPD EMBEDDING_DIM = 768 # 向量维度 # 4. 应用配置 APP_NAME = “tech_support” USER_ID = “user_123” print(f“✓ 使用模型配置：”) print(f“ LLM: {LLM_MODEL}”) print(f“ Embedding: {EMBEDDING_MODEL} （维度： {EMBEDDING_DIM})”) 3.Milvus 数据库初始化

创建向量数据库的 Collection（类似关系数据库的表）

# ==================== 初始化 Milvus ==================== def init_milvus(): “”“初始化 Milvus 连接和 Collection”“” # 第一步：建立连接 try: connections.connect( alias=“default”, host=MILVUS_HOST, port=MILVUS_PORT ) print(f“✓ 已连接到 Milvus: {MILVUS_HOST}:{MILVUS_PORT}”) except Exception as e: print(f“✗ Milvus 连接失败： {e}”) print(“提示：确保 Milvus 已启动”) raise # 第二步：定义数据结构（Schema) fields = [ FieldSchema(name=“id”, dtype=DataType. INT64, is_primary=True, auto_id=True), FieldSchema(name=“user_id”, dtype=DataType. VARCHAR, max_length=100), FieldSchema(name=“session_id”, dtype=DataType. VARCHAR, max_length=100), FieldSchema(name=“question”, dtype=DataType. VARCHAR, max_length=2000), FieldSchema(name=“solution”, dtype=DataType. VARCHAR, max_length=5000), FieldSchema(name=“embedding”, dtype=DataType. FLOAT_VECTOR, dim=EMBEDDING_DIM), FieldSchema(name=“timestamp”, dtype=DataType. INT64) ] schema = CollectionSchema(fields, description=“Tech support memory”) collection_name = “support_memory” # 第三步：创建或获取 Collection if utility.has_collection(collection_name): memory_collection = Collection(name=collection_name) print(f“✓ Collection ‘{collection_name}’ 已存在”) else: memory_collection = Collection(name=collection_name, schema=schema) # 第四步：创建向量索引 index_params = { “index_type”: “IVF_FLAT”, “metric_type”: “COSINE”, “params”: {“nlist”: 128} } memory_collection.create_index(field_name=“embedding”, index_params=index_params) print(f“✓ 已创建 Collection ‘{collection_name}’ 和索引”) return memory_collection # 执行初始化 memory_collection = init_milvus()

4.记忆操作函数实现

封装存储和检索逻辑，作为 Agent 的工具

4.1 存储记忆函数

# ==================== 记忆操作函数 ==================== def store_memory(question: str, solution: str) -> str: “”“ 存储一条解决方案到记忆库 Args: question: 用户的问题 solution: 解决方案 Returns: str: 存储结果消息 ”“” try: print(f“\n【工具调用】 store_memory”) print(f“ - question: {question[:50]}……”) print(f“ - solution: {solution[:50]}……”) # 使用全局 USER_ID（生产环境应从 ToolContext 获取） user_id = USER_ID session_id = f“session_{int(time.time())}” # 关键步骤 1：将问题转换为 768 维向量 embedding_result = genai.embed_content( model=EMBEDDING_MODEL, content=question, task_type=“retrieval_document”,# 指定为文档索引任务 output_dimensionality=EMBEDDING_DIM ) embedding = embedding_result[“embedding”] # 关键步骤 2：插入 Milvus memory_collection.insert([{ “user_id”: user_id, “session_id”: session_id, “question”: question, “solution”: solution, “embedding”: embedding, “timestamp”: int(time.time()) }]) # 关键步骤 3：刷新到磁盘（确保数据持久化） memory_collection.flush() result = f“✓ 已成功存储到记忆库” print(f“【工具结果】 {result}”) return result except Exception as e: error_msg = f“✗ 存储失败： {str(e)}” print(f“【工具错误】 {error_msg}”) return error_msg

4.2 检索记忆函数

def recall_memory(query: str, top_k: int = 3) -> str: “”“ 从记忆库检索相关历史案例 Args: query: 查询问题 top_k: 返回的最相似结果数量 Returns: str: 检索结果 ”“” try: print(f“\n【工具调用】 recall_memory”) print(f“ - query: {query}”) print(f“ - top_k: {top_k}”) user_id = USER_ID # 关键步骤 1：将查询转换为向量 embedding_result = genai.embed_content( model=EMBEDDING_MODEL, content=query, task_type=“retrieval_query”, # 指定为查询任务（和索引时不同） output_dimensionality=EMBEDDING_DIM ) query_embedding = embedding_result[“embedding”] # 关键步骤 2：加载 Collection 到内存（首次查询需要） memory_collection.load() # 关键步骤 3：混合检索（向量相似度 + 标量过滤） results = memory_collection.search( data=[query_embedding], anns_field=“embedding”, param={“metric_type”: “COSINE”, “params”: {“nprobe”: 10}}, limit=top_k, expr=f‘user_id == “{user_id}”’, # 🔑 用户隔离的关键 output_fields=[“question”, “solution”, “timestamp”] ) # 关键步骤 4：格式化结果 if not results[0]: result = “没有找到相关历史案例” print(f“【工具结果】 {result}”) return result result_text = f“找到 {len(results[0])} 个相关案例：\n\n” for i, hit in enumerate(results[0]): result_text += f“案例{i+1}（相似度：{hit.score:.2f}):\n” result_text += f“问题： {hit.entity.get(‘question’)}\n” result_text += f“解决方案： {hit.entity.get(‘solution’)}\n\n” print(f“【工具结果】 找到 {len(results[0])} 个案例”) return result_text except Exception as e: error_msg = f“检索失败： {str(e)}” print(f“【工具错误】 {error_msg}”) return error_msg

4.3 注册为 ADK 工具

#使用 FunctionTool 包装函数 store_memory_tool = FunctionTool(func=store_memory) recall_memory_tool = FunctionTool(func=recall_memory) memory_tools = [store_memory_tool, recall_memory_tool]

5. Agent 定义

核心：定义 Agent 的行为逻辑

# ==================== 创建 Agent ==================== support_agent = Agent( model=LLM_MODEL, name=“support_agent”, description=“技术支持专家 Agent，能够记忆和召回历史案例”, # 关键：Instruction 决定了 Agent 的行为模式 instruction=“”“你是一个技术支持专家。严格按照以下流程处理： <b>当用户提出技术问题时：</b> 1. 立即调用 recall_memory 工具查找历史案例 - 参数 query: 直接使用用户的问题文本 - 不要询问任何其他信息，直接调用 2. 根据检索结果回答： - 如果找到相关案例：说明找到了相似的历史案例，并参考其解决方案回答 - 如果没找到：说明这是新问题，根据你的知识库回答 3. 回答完毕后，询问：”这个方案是否解决了您的问题？“ <b>当用户确认问题已解决时：</b> - 立即调用 store_memory 工具存储这次问答 - 参数 question: 用户最初的问题 - 参数 solution: 你提供的完整解决方案 <b>重要规则：</b> - 必须先调用工具，再回答 - 不要询问 user_id 或任何其他参数 - 只有看到”解决了“、”好了“、”谢谢“等确认词时才存储记忆 ”“”, tools=memory_tools )

6.主程序和运行逻辑

演示跨会话记忆召回的完整流程

# ==================== 主程序 ==================== async def main(): “”“演示跨会话记忆召回”“” # 创建 Session 服务和 Runner session_service = InMemorySessionService() runner = Runner( agent=support_agent, app_name=APP_NAME, session_service=session_service ) # ========== 第一轮对话：建立记忆 ========== print(“\n” + “=”*60) print(“第一轮对话：用户提问并存储方案”) print(“=”*60) session1 = await session_service.create_session( app_name=APP_NAME, user_id=USER_ID, session_id=“session_001” ) # 用户第一次提问 print(“\n【用户】: Milvus 连接超时怎么办？”) content1 = types. Content( role=‘user’, parts=[types. Part(text=“Milvus 连接超时怎么办？”)] ) async for event in runner.run_async( user_id=USER_ID, session_id=session1.id, new_message=content1 ): if event.content and event.content.parts: for part in event.content.parts: if hasattr(part, ‘text’) and part.text: print(f“[Agent]: {part.text}”) # 用户确认问题解决 print(“\n【用户】: 问题解决了，谢谢！”) content2 = types. Content( role=‘user’, parts=[types. Part(text=“问题解决了，谢谢！”)] ) async for event in runner.run_async( user_id=USER_ID, session_id=session1.id, new_message=content2 ): if event.content and event.content.parts: for part in event.content.parts: if hasattr(part, ‘text’) and part.text: print(f“[Agent]: {part.text}”) # ========== 第二轮对话：召回记忆 ========== print(“\n” + “=”*60) print(“第二轮对话：新会话但能召回记忆”) print(“=”*60) session2 = await session_service.create_session( app_name=APP_NAME, user_id=USER_ID, session_id=“session_002” ) # 用户在新会话中提类似问题 print(“\n【用户】: Milvus 连不上了”) content3 = types. Content( role=‘user’, parts=[types. Part(text=“Milvus 连不上了”)] ) async for event in runner.run_async( user_id=USER_ID, session_id=session2.id, new_message=content3 ): if event.content and event.content.parts: for part in event.content.parts: if hasattr(part, ‘text’) and part.text: print(f“[Agent]: {part.text}”) # 程序入口 if __name__ == “__main__”: try: asyncio.run(main()) except KeyboardInterrupt: print(“\n\n 程序已退出”)except Exception as e: print(f“\n\n 程序错误： {e}”) import traceback traceback.print_exc() finally: try: connections.disconnect(alias=“default”) print(“\n✓ 已断开 Milvus 连接”) except: pass

7.运行和测试

7.1 设置环境变量

export GOOGLE_API_KEY="your-gemini-api-key"

python milvus_agent.py

8.预期输出结果

运行结果展示了完整的记忆召回流程：第一轮对话中，Agent 回答了“Milvus 连接超时”的问题并存入记忆库；第二轮启动新会话后，用户换了个说法问“Milvus 连不上了”，向量检索直接匹配到了之前的案例并给出方案。整个过程 Agent 自己决定什么时候该检索历史、什么时候该存储记忆，跨会话召回、语义匹配、用户隔离这几个关键功能。

04

写在最后

ADK 通过 SessionService 和 MemoryService 分离短期上下文和长期知识，Milvus 通过向量检索实现语义匹配和混合查询。

如果你的项目符合“生产环境+工具重度+GCP 基础设施”这三个条件，ADK 值得尝试。如果只是快速验证想法，LangChain 依然是更灵活的选择。

至于记忆系统，不管你用哪个框架，向量数据库都是绕不过去的技术选型。Milvus 的优势在于开源可本地部署、单机支撑千万级向量、支持向量和标量的混合检索，这三点在生产环境中缺一不可。

如何学习大模型 AI ？

由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。

但是具体到个人，只能说是：

“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。

这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。

第一阶段（10天）：初阶应用

该阶段让大家对大模型 AI有一个最前沿的认识，对大模型 AI 的理解超过 95% 的人，可以在相关讨论时发表高级、不跟风、又接地气的见解，别人只会和 AI 聊天，而你能调教 AI，并能用代码将大模型和业务衔接。

• 大模型 AI 能干什么？
• 大模型是怎样获得「智能」的？
• 用好 AI 的核心心法
• 大模型应用业务架构
• 大模型应用技术架构
• 代码示例：向 GPT-3.5 灌入新知识
• 提示工程的意义和核心思想
• Prompt 典型构成
• 指令调优方法论
• 思维链和思维树
• Prompt 攻击和防范
• …

第二阶段（30天）：高阶应用

该阶段我们正式进入大模型 AI 进阶实战学习，学会构造私有知识库，扩展 AI 的能力。快速开发一个完整的基于 agent 对话机器人。掌握功能最强的大模型开发框架，抓住最新的技术进展，适合 Python 和 JavaScript 程序员。

• 为什么要做 RAG
• 搭建一个简单的 ChatPDF
• 检索的基础概念
• 什么是向量表示（Embeddings）
• 向量数据库与向量检索
• 基于向量检索的 RAG
• 搭建 RAG 系统的扩展知识
• 混合检索与 RAG-Fusion 简介
• 向量模型本地部署
• …

第三阶段（30天）：模型训练

恭喜你，如果学到这里，你基本可以找到一份大模型 AI相关的工作，自己也能训练 GPT 了！通过微调，训练自己的垂直大模型，能独立训练开源多模态大模型，掌握更多技术方案。

到此为止，大概2个月的时间。你已经成为了一名“AI小子”。那么你还想往下探索吗？

• 为什么要做 RAG
• 什么是模型
• 什么是模型训练
• 求解器 & 损失函数简介
• 小实验2：手写一个简单的神经网络并训练它
• 什么是训练/预训练/微调/轻量化微调
• Transformer结构简介
• 轻量化微调
• 实验数据集的构建
• …

第四阶段（20天）：商业闭环

对全球大模型从性能、吞吐量、成本等方面有一定的认知，可以在云端和本地等多种环境下部署大模型，找到适合自己的项目/创业方向，做一名被 AI 武装的产品经理。

• 硬件选型
• 带你了解全球大模型
• 使用国产大模型服务
• 搭建 OpenAI 代理
• 热身：基于阿里云 PAI 部署 Stable Diffusion
• 在本地计算机运行大模型
• 大模型的私有化部署
• 基于 vLLM 部署大模型
• 案例：如何优雅地在阿里云私有部署开源大模型
• 部署一套开源 LLM 项目
• 内容安全
• 互联网信息服务算法备案
• …

学习是一个过程，只要学习就会有挑战。天道酬勤，你越努力，就会成为越优秀的自己。

如果你能在15天内完成所有的任务，那你堪称天才。然而，如果你能完成 60-70% 的内容，你就已经开始具备成为一名大模型 AI 的正确特征了。

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】