【Bedrock AgentCore】AI Agent 回答不一致怎么办？双 Memory 架构实现服务标准化（附完整代码）

做过 AI 客服的同学应该都遇到过这个问题：同一个问题，Agent 每次给的答案不完全一样。

我上个月做的航空客服智能体就栽在这上面了。用的 Strands Agent + Bedrock AgentCore，功能都调通了，结果上线第二天业务方就炸了——

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

翻日志才明白，System Prompt 虽然写了补偿政策，但 Agent 推理的时候会根据上下文"灵活调整"。对于标准化服务场景，这种灵活就是事故。

折腾了两天，最后用双 Memory 架构解决了。把方案分享出来，给有类似需求的朋友参考。

问题分析：为什么单 Memory 解决不了

原来的架构只有一个 Memory（下面叫 Memory A），存的是当前用户的对话历史。

Memory A 的作用是保持多轮对话连贯：你说"我要订票"，下一句说"第一个航班"，Agent 能理解"第一个"指前面搜索结果里的第一个。

但 Memory A 是按用户隔离的——Tom 的对话历史只有 Tom 能看到，Lisa 看不到。所以 Agent 处理 Lisa 的问题时，完全不知道之前给 Tom 做过什么方案，每次都从零推理。

根因：缺少跨用户共享的成功经验库。

方案：加一个 Solutions Memory

思路很直接——在 Memory A 之外，加一个 Memory B（Solutions Memory），专门存被用户认可的成功案例：

	Memory A	Memory B
存什么	对话历史	成功案例
谁能看	仅当前用户	所有用户共享
写入方式	Agent 自动写	用户点赞触发
查询方式	actor_id + session_id 精确匹配	语义搜索
数据格式	原始对话	Problem + Solution

两者的协作顺序：

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

AgentCore Memory 底层：STM/LTM 两层架构

Memory A 和 Memory B 底层都用 AgentCore Memory，它有两层存储：

STM（Short-Term Memory） —— 原始事件存储

• 写入后立即可读
• 按 session_id 精确查询
• 存的是完整文本，没有加工

LTM（Long-Term Memory） —— 语义记忆层

• STM 写入后 60-120 秒，后台异步向量化
• 支持语义相似度搜索
• “航班晚点有赔偿吗” 能匹配到 “航班延误 4 小时能给什么补偿”

检索的时候，实际上是 LTM 找到相关记录 → 拿到 session_id → 回 STM 读原文。因为 LTM 里存的是摘要和向量，不是完整原文。

Namespace 分层检索

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)

个人案例优先（更相关），全局案例兜底（新用户也能受益）。

完整代码实现

SolutionsMemory 封装类

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]:
        """分层检索成功案例"""
        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)

        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 _fetch_from_stm(self, actor_id: str, ltm_results: List[Dict]):
        """从 LTM 结果回 STM 拿原文"""
        solutions = []
        for result in ltm_results:
            namespace = result.get("namespaces", [""])[0]
            session_id = namespace.split("/")[4]
            events = self.client.list_events(
                memory_id=self.memory_id,
                actor_id=actor_id,
                session_id=session_id
            )
            if len(events) >= 2:
                solutions.append({
                    "problem": events[0]['content'].replace("Problem: ", ""),
                    "solution": events[1]['content'].replace("Solution: ", ""),
                    "score": result.get("score", 0)
                })
        return solutions

    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

Agent 调用入口

solutions_mem = SolutionsMemory(memory_id="airline_solutions_memory")

async def invoke(user_message: str, actor_id: str, session_id: str):
    # Step 1: Memory B 检索
    cases = solutions_mem.search_solutions(actor_id=actor_id, query=user_message)

    # Step 2: 注入参考案例
    enhanced = user_message
    if cases:
        refs = "\n".join([
            f"问题: {c['problem']}\n方案: {c['solution']}" for c in cases
        ])
        enhanced = f"<reference_solutions>\n{refs}\n</reference_solutions>\n\n用户问题: {user_message}"

    # Step 3: 创建 Agent（Memory A 自动加HG�）
    agent = Agent(
        model="us.anthropic.claude-sonnet-4-20250514",
        system_prompt=SYSTEM_PROMPT,
        memory_id="airline_agent_memory_v2",
        session_id=session_id,
        actor_id=actor_id
    )
    return await agent.invoke(enhanced)

System Prompt 关键部分

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

不加这段，Agent 可能会忽略注入的参考案例。

点赞回调

async def handle_like(actor_id: str, problem: str, solution: str):
    sid = solutions_mem.store_solution(actor_id, problem, solution)
    return {
        "session_id": sid,
        "message": "感谢反馈！此案例将在 1-2 分钟后对其他用户生效。"
    }

踩坑记录

1. LTM 延迟 60-120 秒：案例存储后不能立刻被语义搜索到，测试的时候要等够时间。前端提示"1-2 分钟后生效"。
2. 相似度阈值：设的 0.5，太低会召回不相关案例，太高会漏掉有用的。建议根据业务场景灰度调整。
3. 只存好案例：只有用户点赞才存。不要把所有对话都灌进去，否则坏案例也会被检索到。
4. STM 和 LTM 的分工：LTM 用来"找到相关案例"，STM 用来"读完整原文"。因为 LTM 存的是语义摘要不是原文。

效果

上线后实测：

• Tom 问延误补偿 → Agent 给方案 → Tom 点赞 → 存入 Memory B
• 等 2 分钟
• Lisa 问同样问题 → Memory B 命中 Tom 的案例 → 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
• Strands Agents SDK：https://github.com/strands-agents/sdk-python