AI Agent 总是答不一样？用双 Memory 架构把成功经验沉淀下来

大家好，我是讯享网，很高兴认识大家。这里提供最前沿的Ai技术和互联网信息。

前段时间做一个航空客服智能体，用 Strands Agent + Bedrock AgentCore 搭的。功能跑通了，测试也过了，结果上线第二天就被业务方投诉：

"同样是航班延误 4 小时，昨天给 Tom 的方案是贵宾室 + 200 里程积分，今天给 Lisa 的方案变成了免费改签。Lisa 说她朋友昨天拿到了更好的补偿，凭什么？"

我当时一脸懵——System Prompt 里明明写了补偿政策啊。翻了日志才发现，Agent 在推理的时候会根据上下文"灵活调整"，每次输出的方案细节不完全一致。

对于需要标准化服务的场景来说，这种"灵活"就是 Bug。

我们原来的架构只有一个 Memory（姑且叫 Memory A），它存的是当前用户的对话历史。作用是让多轮对话连贯——你说"我要订票"，下一句说"第一个航班"，Agent 知道"第一个"指的是之前搜索结果里的第一个。

但 Memory A 解决不了跨用户一致性的问题。Tom 的成功对话存在 Tom 的 session 里，Lisa 根本看不到。Agent 每次都是从零开始推理，自然不可能每次输出一样的方案。

想明白这个问题后，思路就清晰了——加一个 Memory B，专门存成功案例，让所有用户共享。

按 actor_id + session_id 精确匹配
管理单个用户的对话上下文
Agent 自动读写，不需要额外代码

用户点赞时才触发存储（不是所有对话都存）
跨用户共享，语义搜索匹配相似问题
格式化为 Problem + Solution 结构

两者协作流程：

用户提问
先查 Memory B——有没有类似的成功案例？
把匹配到的案例注入到 Prompt 里，作为参考
Agent 推理时 Memory A 自动加载对话历史
综合两个 Memory 的信息，生成回复
用户满意点赞 → 存入 Memory B

不管是 Memory A 还是 Memory B，底层都用的 AgentCore Memory 的 STM/LTM 架构：

STM（Short-Term Memory）：写入后立即可读的原始存储。就是个日志数据库，存完整文本，按 session_id 精确查询。

LTM（Long-Term Memory）：写入 STM 后 60-120 秒，AgentCore 后台异步把文本向量化，存到 LTM 里。LTM 支持语义搜索——用户说"航班晚点有赔偿吗"，能匹配到之前存的"航班延误 4 小时能给什么补偿"。

关键点：LTM 有 60-120 秒延迟。这不是 Bug，是异步架构的设计取舍。写入时同步做向量化会拖慢响应，所以选了异步。测试的时候记得等够时间。

Memory B 的数据按 Namespace 组织，支持分层检索：

/solutions/actors/{actor_id}/sessions/{session_id}/

检索策略是先个人后全局：

# Phase 1: 先搜个人案例（个性化优先） namespace = f"/solutions/actors/{actor_id}/sessions/" results = client.retrieve_memories(namespace=namespace, query=query) # Phase 2: 个人没找到，搜全局案例库 if not results: namespace = "/solutions/" results = client.retrieve_memories(namespace=namespace, query=query)

这样设计有三个好处：用户自己的历史案例优先（更相关）、个人案例有一定隐私保护、新用户也能用到全局知识。

from bedrock_agentcore.memory import MemoryClient import uuid from typing import List, Dict class SolutionsMemory: def __init__(self, memory_id: str): self.memory_id = memory_id self.client = MemoryClient() def search_solutions(self, actor_id: str, query: str) -> List[Dict]: """分层检索成功案例""" # Phase 1: 个人案例 personal_ns = f"/solutions/actors/{actor_id}/sessions/" personal = self.client.retrieve_memories( memory_id=self.memory_id, namespace=personal_ns, query=query, top_k=3 ) filtered = [r for r in personal if r.get("score", 0) >= 0.5] if filtered: return self._fetch_from_stm(actor_id, filtered) # Phase 2: 全局案例库 global_results = self.client.retrieve_memories( memory_id=self.memory_id, namespace="/solutions/", query=query, top_k=3 ) filtered = [r for r in global_results if r.get("score", 0) >= 0.5] return self._fetch_from_stm_global(filtered) def store_solution(self, actor_id: str, problem: str, solution: str) -> str: """存储成功案例""" session_id = f"sol-{uuid.uuid4().hex[:8]}" self.client.create_event( memory_id=self.memory_id, actor_id=actor_id, session_id=session_id, messages=[ (f"Problem: {problem}", "USER"), (f"Solution: {solution}", "ASSISTANT") ] ) return session_id

solutions_memory = SolutionsMemory(memory_id="airline_solutions_memory") async def invoke(user_message: str, actor_id: str, session_id: str): # Step 1: 从 Memory B 检索成功案例 similar_cases = solutions_memory.search_solutions( actor_id=actor_id, query=user_message ) # Step 2: 注入到 Prompt enhanced_prompt = user_message if similar_cases: refs = " ".join([ f"问题: {c['problem']} 方案: {c['solution']}" for c in similar_cases ]) enhanced_prompt = f""" 
  
    
    
      {refs} 
     用户问题: {user_message}""" # Step 3: 创建 Agent（Memory A 自动加载对话历史） agent = Agent( model="us.anthropic.claude-sonnet-4-", system_prompt=SYSTEM_PROMPT, memory_id="airline_agent_memory_v2", session_id=session_id, actor_id=actor_id ) response = await agent.invoke(enhanced_prompt) return response

async def handle_like(actor_id: str, problem: str, solution: str): session_id = solutions_memory.store_solution( actor_id=actor_id, problem=problem, solution=solution ) # STM 立即可读，LTM 60-120s 后生效 return {"status": "ok", "session_id": session_id, "message": "感谢反馈！此案例将在 1-2 分钟后对其他用户生效。"}

上线双 Memory 架构后，同样的航班延误场景：

Tom 问延误补偿 → Agent 按政策给出方案 → Tom 满意点赞 → 存入 Memory B
等 2 分钟（LTM 生效）
Lisa 问同样的问题 → Memory B 命中 Tom 的成功案例 → Agent 参考案例给出一致的方案

服务一致性问题解决了。 而且案例库会越滚越大，系统会越来越"懂行"。

Memory B 能发挥作用，关键在 System Prompt 里要明确告诉 Agent 怎么用参考案例：

You are an intelligent customer service agent. When reference solutions are provided in 
  
    
    
      tags: 1. 仔细阅读相似案例和它们的成功处理方式 2. 遵循相同的服务标准和补偿政策 3. 确保所有用户获得一致的服务体验

不加这段的话，Agent 可能会忽略注入的参考案例，又回到"自由发挥"模式。

LTM 延迟：案例存储后 60-120 秒才能被语义搜索到。端到端测试要等够时间。
相似度阈值：我们设的 0.5，太低会召回不相关的案例。建议根据业务场景调整。
案例质量：只有用户主动点赞才存，不要把所有对话都灌进去。
Prompt 注入：成功案例是作为标签注入的，System Prompt 里要明确告诉 Agent 如何使用这些参考。

双 Memory 的核心思路其实很朴素：Memory A 负责”记住这个人”，Memory B 负责”记住好做法”。

如果你也在做需要服务一致性的 Agent 应用——客服、咨询、技术支持这类场景，这个架构值得参考。

完整代码和架构细节可以参考亚马逊云科技官方博客的系列文章：

上篇（基础架构）：https://aws.amazon.com/cn/blogs/china/based-on-aws-bedrock-agentcore-build-enterprise-intelligent-based-on-aidlc-analytics-deploy-practice/
下篇（双 Memory）：https://aws.amazon.com/cn/blogs/china/solutions-memory-ai-agent-case-study-memory-architecture-practice/
AgentCore Memory 文档：https://docs.aws.amazon.com/bedrock/latest/userguide/agentcore-memory.html

AI Agent 总是答不一样？用双 Memory 架构把成功经验沉淀下来

相关推荐