第 7 篇（13-14 章）：AI Agent 的记忆与探索 Microsoft Agent Framework

第 7 篇（13-14 章）：AI Agent 的记忆与探索 Microsoft Agent Framework

本系列「微软《AI Agents for Beginners》实战解读」基于微软官方课程，逐章讲解并结合实践扩展。本篇覆盖第 13 章《AI Agent 的记忆》与第 14 章《探索 Microsoft Agent 框架》。
文中子标题沿用官方课程原文；示例基于 Microsoft Agent Framework（MAF），模型后端为 DeepSeek（OpenAI 兼容协议）。
本篇代码讲解基于真实运行结果（结果已译为中文），每个示例末尾附完整可运行代码。

一、第 13 章：AI Agent 的记忆

AI 代理的独特优势源于两点：调用工具完成任务的能力，以及随时间推移的改进能力。记忆是创建能自我改进、为用户创造更好体验的代理的基础。

1. 理解 AI Agent 记忆

AI Agent 的记忆指允许其保留和回忆信息的机制——对话细节、用户偏好、过往行为、学习到的模式。没有记忆的 AI 应用是无状态的，每次交互从零开始，导致重复且令人沮丧的体验，代理会"忘记"之前的上下文与偏好。

为什么记忆很重要？代理的智能与其回忆并利用过去信息的能力密切相关。记忆使代理能够：反思（从过去行为中学习）、互动（维持持续对话上下文）、主动和反应（基于历史预判需求）、自主（调用存储知识独立操作）。目标是让代理更可靠且有能力。

记忆类型。课程区分多种记忆类型：

• 工作记忆：单次任务或思考过程的"草稿纸"，保存执行下一步所需的即时信息。
• 短期记忆：会话或会话期间的信息，允许代理引用对话中的先前轮次。
• 长期记忆：跨多个会话持续保存的信息（用户偏好、历史交互、一般知识），对个性化至关重要。
• 角色记忆：帮助代理发展一致的"个性"或"角色"。
• 工作流程/情节记忆：保存复杂任务中采取的步骤序列（成功与失败），用于从经验中学习。
• 实体记忆：从对话中提取并记忆具体实体（人物、地点、事物）与事件。
• 结构化 RAG：从各种来源提取密集结构化信息并用以提升回答精度，依赖信息固有结构而非仅语义相似性。

理论扩展：记忆与认知模型。上述分类对应认知科学的三层记忆模型——工作记忆（短暂、容量有限）、短期记忆（会话期）、长期记忆（持久）。工程上，前两者对应会话上下文与线程，后者需外部持久存储（数据库、向量库）并通过工具暴露给代理。

2. 记忆的实现与存储

为代理实现记忆涉及系统化的记忆管理过程：生成、存储、检索、整合、更新甚至"遗忘"。检索尤为关键。

专用记忆工具。

Mem0。作为持久记忆层，使代理回忆相关交互、保存用户偏好，通过两阶段流水线（提取与更新）工作：先总结对话历史并提取新记忆，再基于 LLM 决定添加/修改/删除，存储于向量、图与键值混合数据存储。

Cognee。开源的 AI 代理语义记忆，将结构化与非结构化数据转换为嵌入支持的可查询知识图谱。双存储架构结合向量相似性搜索与图关系，支持混合检索——既理解信息相似性，又理解概念间关联。

理论扩展：检索-增强记忆。长期记忆的工程实现本质是 RAG——将记忆写入向量库，检索时按语义相关度召回并注入上下文（与第 05 章 RAG 管线同构）。Mem0 与 Cognee 的区别在于记忆结构化程度：前者以混合存储为主，后者以知识图谱显式建模实体关系。

3. 使 AI Agent 自我改进

自我改进代理的常见模式是引入"知识代理"：独立代理观察用户与主代理的对话，其职责为：

1. 识别有价值信息：判断对话中是否有值得保存的内容（通用知识或用户偏好）。
2. 提取和总结：萃取关键学习或偏好。
3. 存入知识库：持久化到向量数据库以便后续检索。
4. 增强未来查询：新查询时检索相关内容并附加到提示（类似 RAG）。

理论扩展：延迟与成本优化。自我改进需控制开销：先以廉价快速的模型判断信息是否有存储价值，仅在必要时调用复杂的提取/检索流程；对增长的知识库，可将较少使用的信息迁移至"冷存储"以控制成本。

二、第 14 章：探索 Microsoft Agent 框架

Microsoft Agent Framework（MAF）是微软用于构建 AI 代理的统一框架，兼顾生产与研究场景。

1. 了解 Microsoft Agent 框架

MAF 支持多种编排场景：顺序代理编排（逐步工作流）、并发编排（代理同时完成任务）、群聊编排（代理协同完成单一任务）、交接编排（代理完成子任务后相互交接）、磁性编排（主管代理创建修改任务列表并协调子代理）。

为在生产中交付代理，MAF 提供：可观测性（OpenTelemetry 追踪每个动作）、安全性（角色访问控制、私有数据处理、内容安全）、持久性（线程与工作流可暂停、恢复、从错误恢复）、控制权（人机协同工作流，任务可标记需人工审批）。

MAF 注重互操作性：云无关（容器/本地/多云）、提供商无关（Azure OpenAI、OpenAI 等 SDK）、集成开放标准（A2A、MCP）、插件与连接器（Fabric、SharePoint、Pinecone、Qdrant 等数据与内存服务）。

理论扩展：框架的能力维度。MAF 的能力可归纳为四个维度——编排（组织多代理拓扑）、运维（可观测性、持久性、安全）、互操作（开放标准与跨提供商）、扩展（插件与连接器）。这构成生产级代理框架的完整能力模型。

2. Microsoft Agent 框架核心概念

代理。代理通过定义推理服务（LLM 提供商）、指令与name创建。可用多种服务创建（Azure OpenAI、OpenAI、兼容端点），并支持基于 A2A 协议的远程代理。代理通过.run或.run_stream运行。

工具。工具在定义代理时指定，也可在运行时指定（仅对单次运行生效）。

代理线程。线程处理多轮对话，可通过get_new_thread()创建并被序列化存储、反序列化恢复。

代理中间件。中间件允许在代理与工具/LLM 交互时执行操作：函数中间件（在函数调用前后执行，如日志记录）、聊天中间件（在发送请求给 LLM 前后执行）。

代理记忆。MAF 提供多种记忆：内存存储（应用运行期间线程内）、持久消息（跨会话存储对话历史）、动态记忆（运行前添加到上下文，可存储于 Mem0 等外部服务）。

代理可观测性。MAF 集成 OpenTelemetry，提供跟踪与计量。

工作流。MAF 提供由预定义步骤组成的工作流，以 AI 代理作为组件。核心组件包括执行器（接收输入、执行、输出）与边（定义消息流向——直接边、条件边、开关-条件边、分发边、合并边），并提供内置执行事件（启动、输出、错误、执行器调用/完成等）。

理论扩展：工作流与 DAG。工作流将代理组织为有向无环图（DAG）——执行器为节点，边为数据流。条件边、分发边、合并边分别对应条件路由、扇出、扇入拓扑，构成复杂编排的基础原语。

3. 高级 MAF 模式

构建复杂代理时可考虑的高级模式：中间件组合（链式连接多个中间件，实现日志、认证、限流）、工作流检查点（利用事件与序列化保存/恢复长时运行进程）、动态工具选择（结合工具描述的 RAG 与工具注册，仅展示相关工具）、多代理交接（利用工作流边与条件路由协调专职代理交接）。

4. 在 Microsoft Foundry 上托管 LangChain / LangGraph 代理

MAF 具备框架互操作性——已有用 LangChain/LangGraph 构建的代理可作为 Foundry 托管代理运行，由 Foundry 管理运行时、会话、扩展、身份与协议端点，代理逻辑仍保持于 LangGraph。这通过langchain_azure_ai.agents.hosting包实现，暴露编译后的 LangGraph 图，通过 Foundry 托管代理使用的协议通信。

理论扩展：托管抽象。该能力体现"框架无关"设计——代理的业务逻辑（图、工具、状态）与运行基础设施（托管、身份、协议端点）解耦。开发者可用惯用框架构建逻辑，再交由统一平台承载运维关注点。

三、代码示例讲解：记忆与框架如何真实落地

以下代码取自课程官方示例（已适配 DeepSeek 后端），并附实际运行结果（译为中文）。重点回答：长期记忆如何让代理"跨会话记住用户"？工作流又如何组织多代理协作？

示例一：长期记忆（第 13 章）

① 工作记忆：会话线程

先用会话线程演示短期记忆——同一session传入多次run，代理保持多轮上下文：

session=agent.create_session()print(awaitagent.run("我喜欢海滩目的地，预算3000美元",session=session))print(awaitagent.run("我刚才说的预算是多少？",session=session))

② 实际运行结果（短期记忆，译为中文节选）

（用户：我喜欢海滩目的地，预算3000美元） 代理：已记入你的档案：海滩目的地是你的偏好，总旅行预算 3000 美元。 为了推荐完美海滩，能否帮我缩小范围：出行日期？从哪飞？单人/情侣/家庭？全包度假村还是精品酒店？ （用户：我刚才说的预算是多少？） 代理：你提到本次旅行的预算是 3000 美元。要我据此继续规划，还是有需要调整的？

③ 长期记忆：工具 + 持久存储

短期记忆随会话结束而消失。长期记忆通过工具把偏好写入外部存储，跨会话可用：

preference_store:dict[str,list[str]]={}@tool(approval_mode="never_require")defsave_preference(user_id:Annotated[str,"用户标识"],preference:Annotated[str,"要记住的偏好"])->str:"""保存用户偏好到长期记忆。"""preference_store.setdefault(user_id,[]).append(preference)returnf"已存储：{preference}"@tool(approval_mode="never_require")defget_preferences(user_id:Annotated[str,"用户标识"])->str:"""读取用户已保存的偏好。"""prefs=preference_store.get(user_id,[])return"保存的偏好：\n- "+"\n- ".join(prefs)ifprefselsef"无{user_id}的偏好记录"

④ 实际运行结果（长期记忆，译为中文节选）

（会话1：我是 Sarah，正在规划 10 周年结婚纪念旅行，喜欢浪漫目的地、精致餐饮与水疗， 丈夫行动不便需要无障碍住宿，预算每晚 700-800 美元） 代理：提前祝你周年快乐！已保存你的全部偏好： - 场合：10 周年结婚纪念 - 风格：浪漫目的地 - 餐饮：精致餐饮 - 放松：水疗体验 - 无障碍：需要无障碍住宿（行动需求） - 预算：约 700-800 美元/晚 （会话1续：我们素食，且有严重坚果过敏，请记录） 代理：已记录，Sarah！素食与严重坚果过敏，未来所有推荐都会注意。 （查看存储内容）用户 sarah_johnson_123 的偏好： 10周年结婚纪念 / 浪漫目的地 / 精致餐饮 / 水疗体验 / 无障碍住宿（丈夫行动不便）/ 每晚700-800美元 / 素食 / 严重坚果过敏 （会话2：全新对话线程——"你好，我和丈夫要规划另一趟旅行，能推荐酒店吗？"） 代理：很高兴帮你规划下一次旅行！我从你的档案中看到——这是 10 周年结婚纪念， 你喜欢浪漫目的地、精致餐饮与水疗，我会优先考虑无障碍住宿并控制在 700-800 美元/晚预算内。 请问目的地和日期？ 💡 即使这是全新的对话线程，代理仍从长期记忆取回了 Sarah 的偏好。

⑤ 机制总结：长期记忆如何"跨会话存活"

结合代码与运行结果：

会话1（线程A）→ save_preference 工具 → 偏好写入持久存储 → 会话结束，线程A 短期记忆消失 会话2（全新线程B）→ get_preferences 工具 → 从存储读回偏好 → 注入上下文 → 代理"记得"Sarah，即使线程已换

三个关键机制：

1. 工具即接口：长期记忆不依赖会话，而是通过save_preference/get_preferences工具访问外部存储——工具是代理与持久层之间的桥。
2. 存储独立于会话：偏好存于preference_store（生产为 Mem0/向量库），线程结束不丢失，因此"全新线程仍记得"。
3. 注入时机：系统提示要求代理"会话开始先调get_preferences检查"——记忆在需要时被主动检索并注入上下文（类似 RAG）。

⑥ 完整代码（示例一）

importasynciofromtypingimportAnnotatedfromagent_frameworkimporttoolfrom_deepseekimportmake_client preference_store:dict[str,list[str]]={}@tool(approval_mode="never_require")defsave_preference(user_id:Annotated[str,"用户标识"],preference:Annotated[str,"要记住的偏好"])->str:"""保存用户偏好到长期记忆。"""preference_store.setdefault(user_id,[]).append(preference)returnf"已存储：{preference}"@tool(approval_mode="never_require")defget_preferences(user_id:Annotated[str,"用户标识"])->str:"""读取用户已保存的偏好。"""prefs=preference_store.get(user_id,[])return"保存的偏好：\n- "+"\n- ".join(prefs)ifprefselsef"无{user_id}的偏好记录"asyncdefmain():agent=make_client().as_agent(tools=[save_preference,get_preferences],name="记忆代理",instructions="用户告知偏好时用 save_preference 保存；会话开始先用 get_preferences 检查已存偏好。",)# 会话1：保存 Sarah 的偏好session_1=agent.create_session()print(awaitagent.run("我是Sarah，正在规划10周年结婚纪念旅行，喜欢浪漫目的地、精致餐饮与水疗，丈夫行动不便需无障碍住宿，预算每晚700-800美元。",session=session_1))print(awaitagent.run("我们素食，且有严重坚果过敏，请记录。",session=session_1))# 会话2：全新线程，验证长期记忆session_2=agent.create_session()print(awaitagent.run("你好，我和丈夫要规划另一趟旅行，能推荐酒店吗？",session=session_2))if__name__=="__main__":asyncio.run(main())

示例二：MAF 顺序工作流（第 14 章）

① 用 WorkflowBuilder 串联专职代理

WorkflowBuilder把多个专职代理串成流水线，前一代理输出自动流入后一代理：

fromagent_frameworkimportWorkflowBuilder front_desk=client.as_agent(name="front-desk-agent",instructions="推荐一座城市的最佳景点。")concierge=client.as_agent(name="concierge-agent",instructions="审查景点并给出评分与建议。")workflow=(WorkflowBuilder(start_executor=front_desk).add_edge(front_desk,concierge).build())

② 实际运行观察（工作流执行，译为中文要点）

=== 处理斯德哥尔摩的景点推荐 === [前端代理] 推荐瓦萨博物馆（Vasa Museum）—— 一座海事博物馆，馆内陈列着近乎完整打捞的 17 世纪战舰（1628 年沉没、1961 年打捞）。 [礼宾代理] 审查瓦萨博物馆—— 专业评估：斯德哥尔摩最具标志性的必看景点，战舰的规模与工艺令人叹为观止， 建议至少预留两小时从多个楼层欣赏，清晨到访可避开人流； 备选景点：斯德哥尔摩老城…… === 信息交接 === 流程分析：用户 → 前端代理（推荐）→ 礼宾代理（审查评分），共 2 个代理，信息从前端推荐交接给礼宾输入。

真实运行观察：本次运行中，工作流执行成功（前端代理 → 礼宾代理完成交接与审查），但结构化输出的 Pydantic 校验出现失败——模型输出的 JSON 字段（name/description）与定义的 Schema（attraction_name/city）不匹配。这揭示了真实生产的常见问题：模型并不总能严格遵循输出 Schema，结构化输出需要兜底与容错（对应第 10 章评估与第 12 章上下文验证）。

③ 机制总结：顺序工作流如何组织多代理

结合代码与运行观察：

WorkflowBuilder(start_executor=前端) → add_edge(前端, 礼宾) → build() → 前端输出自动流入礼宾（信息交接） → 两个代理各司其职（推荐 vs 审查），指令聚焦 → 结构化输出需要容错（Schema 可能不被严格遵循）

三个关键机制：

1. 有向图编排：节点为代理、边为数据流，add_edge声明"前端 → 礼宾"，信息自动交接。
2. 职责隔离：前端专注"推荐"，礼宾专注"审查"——每个代理指令简单，质量高于全能代理。
3. 容错意识：真实运行暴露了结构化输出的脆弱性——生产系统必须对 Schema 校验失败设计回退（如重新生成、宽松解析、人工复核）。

④ 完整代码（示例二）

importasynciofromagent_frameworkimportWorkflowBuilderfrom_deepseekimportmake_clientasyncdefmain():client=make_client()front_desk=client.as_agent(name="front-desk-agent",instructions="你是知识渊博的酒店前台，当客人询问某城市景点时，推荐一个热门旅游景点并说明亮点。",)concierge=client.as_agent(name="concierge-agent",instructions="你是精通全球景点的礼宾。对收到的景点推荐给出专家审查与评分，并建议替代景点。",)workflow=(WorkflowBuilder(start_executor=front_desk).add_edge(front_desk,concierge).build())asyncforeventinworkflow.run("I want to visit an attraction in Stockholm",stream=True):ifevent.type=="output":print(getattr(event.data,"text",event.data),end="",flush=True)print()if__name__=="__main__":asyncio.run(main())

运行前提：已按第 1 篇完成环境配置——创建.env（DEEPSEEK_API_KEY填入开发者自有密钥）与_deepseek.py共享客户端，并执行pip install agent-framework python-dotenv openai。