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

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

文章目录

• 第 7 篇（13-14 章）：AI Agent 的记忆与探索 Microsoft Agent Framework
• 一、第 13 章：AI Agent 的记忆
• • 1. 理解 AI Agent 记忆
  • 2. 记忆的实现与存储
  • 3. 使 AI Agent 自我改进
• 二、第 14 章：探索 Microsoft Agent 框架
• • 1. 了解 Microsoft Agent 框架
  • 2. Microsoft Agent 框架核心概念
  • 3. 高级 MAF 模式
  • 4. 在 Microsoft Foundry 上托管 LangChain / LangGraph 代理
• 三、代码示例讲解：记忆与框架如何真实落地
• • 示例一：长期记忆（第 13 章）
  • • ① 工作记忆：会话线程
    • ② 实际运行结果（短期记忆，译为中文节选）
    • ③ 长期记忆：工具 + 持久存储
    • ④ 实际运行结果（长期记忆，译为中文节选）
    • ⑤ 机制总结：长期记忆如何"跨会话存活"
    • ⑥ 完整代码（示例一）
  • 示例二：MAF 顺序工作流（第 14 章）
  • • ① 用 WorkflowBuilder 串联专职代理
    • ② 实际运行观察（工作流执行，译为中文要点）
    • ③ 机制总结：顺序工作流如何组织多代理
    • ④ 完整代码（示例二）

本系列「微软《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 自我改进

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

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

二、第 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(await agent.run("我喜欢海滩目的地，预算3000美元", session=session))
print(await agent.run("我刚才说的预算是多少？", session=session))

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

（用户：我喜欢海滩目的地，预算3000美元）
代理：已记入你的档案：海滩目的地是你的偏好，总旅行预算 3000 美元。
为了推荐完美海滩，能否帮我缩小范围：出行日期？从哪飞？单人/情侣/家庭？全包度假村还是精品酒店？

（用户：我刚才说的预算是多少？）
代理：你提到本次旅行的预算是 3000 美元。要我据此继续规划，还是有需要调整的？

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

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

preference_store: dict[str, list[str]] = {}

@tool(approval_mode="never_require")
def save_preference(user_id: Annotated[str, "用户标识"], preference: Annotated[str, "要记住的偏好"]) –> str:
"""保存用户偏好到长期记忆。"""
preference_store.setdefault(user_id, []).append(preference)
return f"已存储：{preference}"

@tool(approval_mode="never_require")
def get_preferences(user_id: Annotated[str, "用户标识"]) –> str:
"""读取用户已保存的偏好。"""
prefs = preference_store.get(user_id, [])
return "保存的偏好：\\n- " + "\\n- ".join(prefs) if prefs else f"无 {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，即使线程已换

三个关键机制：

⑥ 完整代码（示例一）

import asyncio
from typing import Annotated
from agent_framework import tool
from _deepseek import make_client

preference_store: dict[str, list[str]] = {}

@tool(approval_mode="never_require")
def save_preference(user_id: Annotated[str, "用户标识"], preference: Annotated[str, "要记住的偏好"]) –> str:
"""保存用户偏好到长期记忆。"""
preference_store.setdefault(user_id, []).append(preference)
return f"已存储：{preference}"

@tool(approval_mode="never_require")
def get_preferences(user_id: Annotated[str, "用户标识"]) –> str:
"""读取用户已保存的偏好。"""
prefs = preference_store.get(user_id, [])
return "保存的偏好：\\n- " + "\\n- ".join(prefs) if prefs else f"无 {user_id} 的偏好记录"

async def main():
agent = make_client().as_agent(
tools=[save_preference, get_preferences],
name="记忆代理",
instructions="用户告知偏好时用 save_preference 保存；会话开始先用 get_preferences 检查已存偏好。",
)
# 会话1：保存 Sarah 的偏好
session_1 = agent.create_session()
print(await agent.run("我是Sarah，正在规划10周年结婚纪念旅行，喜欢浪漫目的地、精致餐饮与水疗，丈夫行动不便需无障碍住宿，预算每晚700-800美元。", session=session_1))
print(await agent.run("我们素食，且有严重坚果过敏，请记录。", session=session_1))

# 会话2：全新线程，验证长期记忆
session_2 = agent.create_session()
print(await agent.run("你好，我和丈夫要规划另一趟旅行，能推荐酒店吗？", session=session_2))

if __name__ == "__main__":
asyncio.run(main())

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

① 用 WorkflowBuilder 串联专职代理

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

from agent_framework import WorkflowBuilder

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 可能不被严格遵循）

三个关键机制：

④ 完整代码（示例二）

import asyncio
from agent_framework import WorkflowBuilder
from _deepseek import make_client

async def main():
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()
)

async for event in workflow.run("I want to visit an attraction in Stockholm", stream=True):
if event.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。