快时尚电商行业智能体设计思路与应用实践（七）Amazon Bedrock AgentCore Runtime 深度解析和场景分析

                              +----------------------------------------+
                              |                                        |
                              |             AWS EC2 INSTANCE           |
        +----------+          | +---------+  +-----------------------+ |
        |          |          | |         |  |  microVM (Container)  | |---+
        |          |          | |         |  |  [ Docker Image ]     | |   |
        |          |          | |         |  +-----------------------+ |   |
        |          |  User 1  | |         |                            |   | microVMs
        |          |--------->| |         |  +-----------------------+ |   | stay
Users   |          |  User 2  | |         |  |  microVM (Container)  | |   | alive
&       | ENDPOINT |--------->| | GATEWAY |  |  [ Docker Image ]     | |   | until
Traffic |          |  User 3  | |         |  +-----------------------+ |   | session
        |          |--------->| |         |                            |   | timeout
        |          |          | |         |  +-----------------------+ |   |
        |          |          | |         |  |  microVM (Container)  | |   |
        |          |          | |         |  |  [ Docker Image ]     | |---+
        +----------+          | +---------+  +-----------------------+ |
                              |                                        |
                              +----------------------------------------+
                              

from strands import Agent, tool
from strands_tools import calculator  # 导入计算器工具，用于计算商品折扣价格
import argparse
import json
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from strands.models import BedrockModel

app = BedrockAgentCoreApp()

# 创建自定义工具：查询商品库存
@tool
def check_inventory(product_id: str):
    """ 查询指定商品的实时库存状态 """
    # 实际实现中会连接库存管理系统
    return {"product_id": product_id, "stock": 150, "warehouse": "华东仓"}

# 创建自定义工具：获取穿搭推荐
@tool
def get_outfit_recommendation(style: str, season: str):
    """ 根据风格和季节获取穿搭推荐 """
    # 实际实现中会调用推荐引擎
    return f"为您推荐{season}季{style}风格的热门单品组合"

# 创建自定义工具：查询物流状态
@tool
def track_shipment(order_id: str):
    """ 查询订单物流配送状态 """
    # 实际实现中会调用物流API
    return {"order_id": order_id, "status": "配送中", "eta": "明天送达"}

model_id = "global.anthropic.claude-sonnet-4-5-20250929-v1:0"
model = BedrockModel(model_id=model_id,)

# 创建快时尚电商智能客服Agent
agent = Agent(
    model=model,
    tools=[calculator, check_inventory, get_outfit_recommendation, track_shipment],
    system_prompt="你是一个专业的快时尚电商智能助手。你可以帮助顾客查询商品库存、计算折扣价格、提供穿搭推荐、追踪物流状态，以及处理退换货咨询。请用友好专业的语气与顾客交流。"
)

@app.entrypoint
def fashion_ecommerce_agent(payload):
    """处理来自顾客的咨询请求"""
    user_input = payload.get("prompt")
    print("顾客咨询:", user_input)
    response = agent(user_input)
    return response.message['content'][0]['text']

if name == "main":
    app.run()

+--------------------------------------------------------------+
| USER                                                         |
+----+---------------------------------------------------------+
     |                              ^
     | (1) User question            | (4) Agent response
     v                              |
+-----------------------------------+--------------------------+
| Local Environment or AWS Cloud (Strands Agents)              |
|                                                              |
|   +-----------+            (2) task            +----------+  |
|   |           | <----------------------------> |  Tool 1  |  |
|   |   AGENT   |                                |  Tool 2  |  |
|   |           |                                |  Tool 3  |  |
|   +-----+-----+                                +----------+  |
|         ^                                                    |
+---------|----------------------------------------------------+
          |
          | (3) Invokes LLM and Processes Outputs
          v
+--------------------------------------------------------------+
| AWS Cloud                                                    |
|                                                              |
|                   +-----------------------+                  |
|                   |  Amazon Bedrock LLMs  |                  |
|                   +-----------------------+                  |
|                                                              |
+--------------------------------------------------------------+

Content-Type: application/json
{
    "prompt": "这款连衣裙还有M码吗？"
}

Content-Type: application/json
{
    "response": "完整响应",
    "status": "success"
}

Content-Type: text/event-stream
data: {"event": "部分响应 1"}
data: {"event": "部分响应 2"}
data: {"event": "部分响应 3"}
data: {"event": "最终响应"}

{
    "status": "<status_value>",
    "time_of_last_update": <unix_timestamp>
}

from mcp.server.fastmcp import FastMCP
from starlette.responses import JSONResponse

mcp = FastMCP(host="0.0.0.0", stateless_http=True)

@mcp.tool()
def calculate_discount(original_price: float, discount_rate: float) -> float:
    """计算商品折扣后的价格，用于促销活动"""
    return original_price * (1 - discount_rate)

@mcp.tool()
def check_size_availability(product_id: str, size: str) -> dict:
    """查询指定商品特定尺码的库存情况"""
    # 实际实现中会查询库存数据库
    return {"product_id": product_id, "size": size, "available": True, "quantity": 25}

@mcp.tool()
def get_style_recommendation(customer_id: str, occasion: str) -> str:
    """根据顾客ID和场合获取个性化穿搭推荐"""
    return f"根据您的购买历史，为{occasion}场合推荐：简约风格白色衬衫搭配高腰阔腿裤"

@mcp.tool()
def process_return_request(order_id: str, reason: str) -> dict:
    """处理顾客的退换货申请"""
    return {"order_id": order_id, "status": "已受理", "return_label": "RL-2024-001234"}

if name == "main":
    mcp.run(transport="streamable-http")

{
  "jsonrpc": "2.0",
  "id": "req-001",
  "method": "message/send",
  "params": {
    "message": {
      "role": "user",
      "parts": [
        { "kind": "text", "text": "你好!" }
      ]
    }
  }
}

[ 编写 A2A Server 代码 ]
            ⬇
[ 本地启动 A2A Server (JSON-RPC/9000) ]
            ⬇
[ 打包为容器镜像并上传 ECR ]
            ⬇
[ 使用 AgentCore CLI 部署到 Bedrock AgentCore Runtime ]
            ⬇
[ 生成 & 获取 Agent Card / 部署 Endpoint ]

                          +---------------------------+
                          |      User Interface       |
                          |  (Mobile App / Web Site)  |
                          +-------------+-------------+
                                        |
                                        v
+----------------------------------------------------------------------------------+
|                            Bedrock AgentCore Runtime                             |
|                                                                                  |
|   +--------------------------+   (A2A)        +------------------------------+   |
|   |   Orchestrator Agent     |<-------------->|     Inventory Agent          |   |
|   |   (The "Brain")          |   (A2A)        | (Check SKU, Stock, Warehouse)|   |
|   +------------+-------------+                +------------------------------+   |
|                ^                                                                 |
|                | (A2A)                        +------------------------------+   |
|                +----------------------------->|      Trend/Style Agent       |   |
|                |                              | (Social Media Trends, Style) |   |
|                | (A2A)                        +------------------------------+   |
|                +----------------------------->|      Visual/Outfit Agent     |   |
|                                               | (Virtual Try-on, Mix & Match)|   |
|                                               +------------------------------+   |
|                                                                                  |
+---------------------------------------+------------------------------------------+
                                        |                               
                                        v                                     
+----------------------------------------------------------------------------------+
|  Shared Infrastructure (AWS)                                                     |
 |                                                                                  |
|  +-----------------------+  +-----------------------+  +----------------------+  |
|  |    Cognito (Auth)     |  |   Bedrock (Models)    |  |    External Tools    |  |
|  |  - Token Exchange     |  |  - Nova / Claude      |  |  - ERP / CRM System  |  |
|  +-----------------------+  +-----------------------+  +----------------------+  |
|                                                                                  |
+----------------------------------------------------------------------------------+

                    +--------------------------------------+
                    |                                      |    data stored
      session-id=1  |        +-----------------------+     |    per single session 
    +-------------->|        |         microVM       |     |    +------------------+
    |               | +----->|                       +-----|--->|  (O) Disk        |
    |               | |      |                       +-----|--->| [|||] Memory     |
    |               | |      +-----------------------+     |    +------------------+
    |               | |                                    |
+--------+      +---+-+----+                               |
| User-1 |----->| Endpoint |                               |
+--------+      +---+-+----+                               |
    |               | |                                    |    data stored
    |               | |      +-----------------------+     |    per single session
    | session-id=2  | |      |         microVM       |     |    +------------------+
    +-------------->| +----->|                       +-----|--->|  (O) Disk        |
                    |        |                       +-----|--->| [|||] Memory     |
                    |        +-----------------------+     |    +------------------+
                    |                     |                |
                    +---------------------|----------------+
                                          |
                                          v
                       +----------------------------------+
                       | microVM lives to serve a single  |
                       | session only, and contain the    |
                       | conversation history as long as  |
                       | the session lives                |
                       +----------------------------------+

# 顾客咨询的第一条消息：询问商品信息
response1 = agent_core_client.InvokeAgentRuntime(
    agentRuntimeArn=agent_arn,
    runtimeSessionId="customer-123456-session-123456789",
    payload=json.dumps({
        "prompt": "这款碎花连衣裙有什么颜色可选？"
    }).encode()
)

# 同一对话中的后续消息重用 runtimeSessionId
# 顾客继续询问尺码，Agent能够理解上下文
response2 = agent_core_client.InvokeAgentRuntime(
    agentRuntimeArn=agent_arn,
    runtimeSessionId="customer-123456-session-123456789",
    payload=json.dumps({
        "prompt": "M码还有库存吗？"
    }).encode()
)

# 顾客决定下单，Agent仍然记得之前讨论的商品
response3 = agent_core_client.InvokeAgentRuntime(
    agentRuntimeArn=agent_arn,
    runtimeSessionId="customer-123456-session-123456789",
    payload=json.dumps({
        "prompt": "好的，帮我加入购物车"
    }).encode()
)

# 自动跟踪异步函数：用于处理大批量订单或库存同步等耗时操作
@app.async_task
async def sync_inventory_from_supplier():
    """从供应商系统同步最新库存数据，这是一个耗时操作"""
    await asyncio.sleep(30)  # 状态变为 "HealthyBusy"
    # 实际实现中会调用供应商API同步库存
    return "库存同步完成"

@app.async_task
async def generate_sales_report(date_range: str):
    """生成指定时间段的销售报表，需要处理大量数据"""
    await asyncio.sleep(60)  # 状态变为 "HealthyBusy"
    # 实际实现中会查询数据库并生成报表
    return f"{date_range} 销售报表已生成"

@app.entrypoint
async def handler(event):
    """处理顾客请求，对于耗时操作启动后台任务"""
    task_type = event.get("task_type")
    if task_type == "inventory_sync":
        asyncio.create_task(sync_inventory_from_supplier())
        return {"status": "库存同步任务已启动，请稍后查看结果"}
    elif task_type == "sales_report":
        asyncio.create_task(generate_sales_report(event.get("date_range")))
        return {"status": "报表生成任务已启动，完成后将通知您"}
    return {"status": "未知任务类型"}

[ User ]--------------------> [ Agent ]
        Conversation Messages     |
                                  | Events To Memory
                                  v
                        +-----------------------+
                        |  Short-Term Memory    |
                        |  Conversation Storage |
                        +-----------------------+
                                    | ^
                                    | | 
                                    v |
                  +-----------------------------------------+
                  |         Background Job (Three Steps)    |
                  |                                         |
                  |   1. [ Analyze Conversation ]           |
                  |             |                           | 
                  |             v                           |
                  |   2. [ Information Extraction ]         |
                  |             |                           |
                  |             v                           |
                  |   3. [ Writer To Long-Term Memory ]     |
                  +-----------------------------------------+
                                |
                                |
                                v
                        +-----------------------+
                        |   Long-Term Memory    | 
                        |                       |      
                        +-----------------------+
                                  ^
                                  | Semantic Retrieval
                                  |
                              [ Agent ]

class MemoryHookProvider(HookProvider):
    def __init__(self, memory_client: MemoryClient, memory_id: str):
        self.memory_client = memory_client
        self.memory_id = memory_id

    def on_agent_initialized(self, event: AgentInitializedEvent):
        """当快时尚电商客服智能体启动时加载最近的顾客对话历史"""
        try:
            # 从智能体状态中获取会话信息（如顾客ID、购物会话ID）
            actor_id = event.agent.state.get("actor_id")
            session_id = event.agent.state.get("session_id")

            if not actor_id or not session_id:
                logger.warning("智能体状态中缺少 actor_id 或 session_id")
                return

            # 从记忆中加载最近5轮对话（如顾客之前咨询的尺码、颜色偏好等）
            recent_turns = self.memory_client.get_last_k_turns(
                memory_id=self.memory_id,
                actor_id=actor_id,
                session_id=session_id,
                k=5
            )

            if recent_turns:
                # 格式化对话历史作为上下文
                context_messages = []
                for turn in recent_turns:
                    for message in turn:
                        role = message['role']
                        content = message['content']['text']
                        context_messages.append(f"{role}: {content}")

                context = "\n".join(context_messages)
                # 将上下文添加到智能体的系统提示中，帮助客服了解顾客之前的咨询内容
                event.agent.system_prompt += f"\n\n最近对话记录:\n{context}"
                logger.info(f"✅ 已加载 {len(recent_turns)} 轮对话历史")

        except Exception as e:
            logger.error(f"记忆加载错误: {e}")

    def on_message_added(self, event: MessageAddedEvent):
        """将顾客与客服的对话消息存储到记忆中"""
        messages = event.agent.messages
        try:
            # 从智能体状态中获取会话信息
            actor_id = event.agent.state.get("actor_id")
            session_id = event.agent.state.get("session_id")

            if messages[-1]["content"][0].get("text"):
                # 保存对话消息（如顾客询问的服装款式、尺码、物流等信息）
                self.memory_client.create_event(
                    memory_id=self.memory_id,
                    actor_id=actor_id,
                    session_id=session_id,
                    messages=[(messages[-1]["content"][0]["text"], messages[-1]["role"])]
                )
        except Exception as e:
            logger.error(f"记忆保存错误: {e}")

    def register_hooks(self, registry: HookRegistry):
        # 注册记忆钩子，实现快时尚电商场景下的自动记忆管理
        registry.add_callback(MessageAddedEvent, self.on_message_added)
        registry.add_callback(AgentInitializedEvent, self.on_agent_initialized)

def create_personal_agent():
    """创建具有记忆和商品搜索能力的快时尚电商客服智能体"""
    agent = Agent(
        name="FastFashionAssistant",
        model="global.anthropic.claude-sonnet-4-5-20250929-v1:0",  # 或您偏好的模型
        system_prompt=f"""您是一位专业的快时尚电商客服助手，具备商品搜索和推荐能力。

您可以帮助顾客：
- 查询服装款式、尺码、颜色和库存信息
- 搜索当季流行单品和新品上架信息
- 处理订单查询、物流追踪和退换货咨询
- 根据顾客偏好提供个性化穿搭建议

当您需要查询最新商品信息或流行趋势时，请使用商品搜索功能。
今日日期: {datetime.today().strftime('%Y-%m-%d')}
请保持友好、专业的服务态度，帮助顾客找到心仪的时尚单品。""",
        hooks=[MemoryHookProvider(client, memory_id)],
        tools=[websearch],
        state={"actor_id": ACTOR_ID, "session_id": SESSION_ID}
    )
    return agent

# 创建快时尚电商客服智能体
agent = create_personal_agent()
logger.info("✅ 已创建具有记忆和商品搜索能力的快时尚电商客服智能体")

from bedrock_agentcore.memory import MemoryClient
from bedrock_agentcore.memory.constants import StrategyType

# 初始化 Memory 客户端
client = MemoryClient(region_name=REGION)
memory_name = "FastFashionAgentMemory"

# 为快时尚电商客服定义记忆策略
strategies = [
    {
        StrategyType.USER_PREFERENCE.value: {
            "name": "CustomerFashionPreferences",
            "description": "捕获顾客的时尚偏好和购物行为（如偏好的风格、常购尺码、喜欢的颜色等）",
            "namespaces": ["fashion/customer/{actorId}/preferences"]
        }
    },
    {
        StrategyType.SEMANTIC.value: {
            "name": "CustomerFashionSemantic",
            "description": "存储对话中提取的事实信息（如顾客提到的身材特征、穿搭场景需求等）",
            "namespaces": ["fashion/customer/{actorId}/semantic"],
        }
    }
]

# 创建记忆资源
try:
    memory = client.create_memory_and_wait(
        name=memory_name,
        strategies=strategies,
        # 定义记忆策略，用于存储顾客的时尚偏好和购物历史
        description="快时尚电商客服智能体的记忆存储",
        event_expiry_days=90,  # 记忆在90天后过期，适合快时尚行业的季节性特点
    )
    memory_id = memory['id']
    logger.info(f"✅ 已创建记忆: {memory_id}")
except ClientError as e:
    if e.response['Error']['Code'] == 'ValidationException' and "already exists" in str(e):
        # 如果记忆已存在，获取其ID
        memories = client.list_memories()
        memory_id = next((m['id'] for m in memories if m['id'].startswith(memory_name)), None)
        logger.info(f"记忆已存在，使用现有记忆ID: {memory_id}")
except Exception as e:
    # 处理记忆创建过程中的任何错误
    logger.info(f"❌ 错误: {e}")
    import traceback
    traceback.print_exc()
    # 出错时清理 - 如果记忆已部分创建则删除
    if memory_id:
        try:
            client.delete_memory_and_wait(memoryId=memory_id, max_wait=300)
            logger.info(f"已清理记忆: {memory_id}")
        except Exception as cleanup_error:
            logger.info(f"清理记忆失败: {cleanup_error}")

class MemoryHookProvider(HookProvider):
    """用于自动记忆管理的钩子提供者，适用于快时尚电商场景"""

    def __init__(self, memory_id: str, client: MemoryClient):
        self.memory_id = memory_id
        self.client = client

    def retrieve_memories(self, event: MessageAddedEvent):
        """在处理顾客消息前检索相关记忆（如之前咨询过的款式、尺码偏好等）"""
        messages = event.agent.messages
        if messages[-1]["role"] == "user" and "toolResult" not in messages[-1]["content"][0]:
            user_message = messages[-1]["content"][0].get("text", "")

            try:
                # 从智能体状态获取顾客ID
                actor_id = event.agent.state.get("actor_id")
                if not actor_id:
                    logger.warning("智能体状态中缺少 actor_id")
                    return

                # 构建命名空间路径，用于检索该顾客的时尚偏好记忆
                namespace = f"/fashion/customer/{actor_id}"

                # 检索与当前查询相关的记忆（如顾客之前表达的风格偏好、尺码信息等）
                memories = self.client.retrieve_memories(
                    memory_id=self.memory_id,
                    namespace=namespace,
                    query=user_message
                )

                # 提取记忆内容
                memory_context = []
                for memory in memories:
                    if isinstance(memory, dict):
                        content = memory.get('content', {})
                        if isinstance(content, dict):
                            text = content.get('text', '').strip()
                            if text:
                                memory_context.append(text)

                # 将记忆注入到用户消息中，帮助客服了解顾客的历史偏好
                if memory_context:
                    context_text = "\n".join(memory_context)
                    original_text = messages[-1]["content"][0].get("text", "")
                    messages[-1]["content"][0]["text"] = (
                        f"{original_text}\n\n顾客历史偏好: {context_text}"
                    )
                    logger.info(f"已检索 {len(memory_context)} 条相关记忆")

            except Exception as e:
                logger.error(f"检索记忆失败: {e}")

    def save_memories(self, event: AfterInvocationEvent):
        """在智能体响应后保存对话内容（记录顾客的咨询内容和客服的推荐建议）"""
        try:
            messages = event.agent.messages
            if len(messages) >= 2 and messages[-1]["role"] == "assistant":
                # 获取最后的用户消息和助手消息
                user_msg = None
                assistant_msg = None

                for msg in reversed(messages):
                    if msg["role"] == "assistant" and not assistant_msg:
                        assistant_msg = msg["content"][0]["text"]
                    elif msg["role"] == "user" and not user_msg and "toolResult" not in msg["content"][0]:
                        user_msg = msg["content"][0]["text"]
                        break

                if user_msg and assistant_msg:
                    # 从智能体状态获取会话信息
                    actor_id = event.agent.state.get("actor_id")
                    session_id = event.agent.state.get("session_id")

                    if not actor_id or not session_id:
                        logger.warning("智能体状态中缺少 actor_id 或 session_id")
                        return

                    # 保存对话到记忆（包含顾客咨询和客服推荐的服装信息）
                    self.client.create_event(
                        memory_id=self.memory_id,
                        actor_id=actor_id,
                        session_id=session_id,
                        messages=[(user_msg, "USER"), (assistant_msg, "ASSISTANT")]
                    )
                    logger.info("已将对话保存到记忆")

        except Exception as e:
            logger.error(f"保存记忆失败: {e}")

    def register_hooks(self, registry: HookRegistry) -> None:
        """注册记忆钩子，实现快时尚电商场景下的自动化记忆管理"""
        registry.add_callback(MessageAddedEvent, self.retrieve_memories)
        registry.add_callback(AfterInvocationEvent, self.save_memories)
        logger.info("记忆钩子已注册")

 +---------------------------------------------------------------+ 
 |                    AgentCore Runtime Layer                    | 
 |                                                               | 
 |   session_id                                                  | 
 |   (single conversation / single reasoning lifecycle)          | 
 |                                                               | 
 |   +---------------+   +---------------+   +---------------+   | 
 |   |   Agent A     |   |   Agent B     |   |   Agent C     |   | 
 |   |               |   |               |   |               |   | 
 |   |  STM / Msgs   |   |  STM / Msgs   |   |  STM / Msgs   |   | 
 |   |  Hooks        |   |  Hooks        |   |  Hooks        |   | 
 |   |               |   |               |   |               |   | 
 |   |  memory_id    |   |  memory_id    |   |  memory_id    |   | 
 |   |  branch = A   |   |  branch = B   |   |  branch = C   |   | 
 |   |               |   |               |   |               |   | 
 |   +---------------+   +---------------+   +---------------+   | 
 |                                                               | 
 |                                                               | 
 +---------------------------------------------------------------+ 
                       |                                           
                       |  (persistent across sessions)             
                       v                                           
 +---------------------------------------------------------------+ 
 |                Organized Storage / LTM Layer                  | 
 |                                                               | 
 |   memory_id                                                   | 
 |   (long-term memory namespace)                                | 
 |                                                               | 
 |   +-------------------------------------------------------+   | 
 |   |  Branch: A   ← written/read by Agent A                |   | 
 |   +-------------------------------------------------------+   | 
 |                                                               | 
 |   +-------------------------------------------------------+   | 
 |   |  Branch: B   ← written/read by Agent B                |   | 
 |   +-------------------------------------------------------+   | 
 |                                                               | 
 |   +-------------------------------------------------------+   | 
 |   |  Branch: C   ← written/read by Agent C                |   | 
 |   +-------------------------------------------------------+   | 
 |                                                               | 
 +---------------------------------------------------------------+