制造业智能化转型新引擎：基于AWS Bedrock AgentCore构建生产管理智能体系统

from bedrock_agentcore.memory import MemoryClient

# 初始化Memory客户端
client = MemoryClient(region_name="us-west-2")

# 创建短期记忆资源
memory = client.create_memory(
    name="ManufacturingAgentMemory",
    description="制造业智能体的会话记忆管理",
)

print(f"Memory ID: {memory.get('id')}")

# 存储生产交互事件到短期记忆 - 展示问题解决和经验积累过程
client.create_event(
    memory_id=memory.get("id"),
    actor_id="ProductionOperator01",  # 操作员ID
    session_id="ProductionSession_20241201_001",  # 生产会话ID
    messages=[
        ("设备A生产线温度异常，当前读数85°C，已超过标准阈值80°C。请立即协助排查原因。", "USER"),
        ("收到。正在查询历史数据库中关于'设备A温度超标'的类似问题解决方案...", "ASSISTANT"),
        ("query_historical_solutions(issue_type='temperature_anomaly', equipment='A')", "TOOL"),
        ("查询到3条相关历史记录：\n1. 记录#1023：冷却液液位过低，补充后解决。\n2. 记录#1057：冷却风扇故障，更换后解决。\n3. 记录#1101：热交换器堵塞，清理后解决。", "TOOL"),
        ("历史数据显示，最常见的三个原因是：冷却液不足、风扇故障或热交换器堵塞。建议首先检查冷却液液位，这是最快、最简单的排查步骤。", "ASSISTANT"),
        ("有道理。不过我感觉车间环境温度好像有点高。最好也同步检查一下当前的环境温度和通风系统的状态。", "USER"),
        ("很好的补充建议，经验非常宝贵。正在调用传感器读取环境数据...", "ASSISTANT"),
        ("check_environmental_factors(equipment='A', factors=['ambient_temperature', 'ventilation_status'])", "TOOL"),
        ("环境温度: 32.1°C。\n通风系统状态: 运行中。\n出风口风速: 5.2 m/s。", "TOOL"),
        ("收到环境数据。当前环境温度32.1°C，高于28°C的标准上限。通风系统出风口风速为5.2 m/s，根据设备手册，这低于正常运行所需的7.5 m/s标准，风量明显不足，通常由过滤网堵塞引起。\n\n**综合诊断更新**：\n1. **主要外部因素**：已确认**环境温度过高**和**通风系统风量不足**。\n2. **待查内部因素**：根据历史记录，仍需排查冷却液、风扇或热交换器问题。\n\n**建议操作**：\n- **紧急处理**：立即派人清理通风系统过滤网，并增加车间外部冷风补给。\n- **同步检查**：同时，请现场人员检查冷却液液位是否正常。", "ASSISTANT"),
        ("很好，下次遇到这种情况的时候，记得将‘高温天气下，设备温度异常需同时排查环境因素和通风系统’这一复合诊断策略，标记为重要知识点，并关联到‘设备A温度异常’的解决方案中。", "USER"),
        ("收到。下次遇到类似情况的时候将同步检查环境温度和通风系统状态。", "ASSISTANT")
    ],
)

通过查询经过此前短时记忆提取到的内容，可以看到，Memory 模块从中提取到了关键信息：
Long term memory: [{
  "memoryRecordId": "mem-0fc4d8f8d06e51138bde67747301166dd611_1",
  "content": {
    "text": "    <topic name=\"设备A温度异常情况\">\n        设备A生产线温度读数85°C，已超过标准阈值80°C。\n    </topic>\n    <topic name=\"历史记录查询\">\n        系统查询到3条相关历史记录：\n        1. 冷却液液位过低\n        2. 冷却风扇故障\n        3. 热交换器堵塞\n    </topic>\n    <topic name=\"环境数据检查\">\n        环境温度32.1°C，高于28°C的标准上限。\n        通风系统出风口风速为5.2 m/s，低于正常运行所需的7.5 m/s标准。\n    </topic>\n    <topic name=\"诊断结果\">\n        主要外部因素：环境温度过高和通风系统风量不足。\n        待查内部因素：冷却液、风扇或热交换器问题。\n    </topic>\n    <topic name=\"建议操作\">\n        紧急处理：清理通风系统过滤网，增加车间外部冷风补给。\n        同步检查：检查冷却液液位是否正常。\n    </topic>\n    <topic name=\"用户反馈\">\n        用户要求在未来类似情况下，除了参考知识库方案外，也要提醒关注环境温度因素。\n    </topic>\n"
  },
  "memoryStrategyId": "SessionSummarizer-pmJRWvDiUc",
  "namespaces": [
    "/summaries/user_example1/session_example1"
  ],
  "createdAt": "2025-09-29T11:51:05+08:00",
  "score": 0.39427978
}]

首先需要初始化Gateway客户端和相关的AWS服务连接：
from bedrock_agentcore_starter_toolkit.operations.gateway.client import GatewayClient
import boto3

class ManufacturingGatewaySetup:
    def __init__(self, region="us-west-2"):
        self.region = region
        self.client = GatewayClient(region_name=region)
        self.boto_client = boto3.client("bedrock-agentcore-control", region_name=region)

def setup_manufacturing_gateway(self, gateway_name="ManufacturingGateway"):
    # 创建OAuth授权服务器
    cognito_response = self.client.create_oauth_authorizer_with_cognito(gateway_name)
    
    # 保存认证配置供后续使用
    with open(f'gateway/{gateway_name.lower()}_cognito_response.json', 'w') as outfile:
        json.dump(cognito_response, outfile, indent=2)

gateway = self.client.create_mcp_gateway(
    name=f"{gateway_name}-{int(time.time())}",  # 时间戳确保唯一性
    role_arn=None,  # 自动创建IAM角色
    authorizer_config=cognito_response["authorizer_config"],
    enable_semantic_search=True,  # 启用语义搜索功能
)

# IAM权限传播等待
time.sleep(30)

def add_manufacturing_api_target(self, gateway, api_schema_file, api_key, target_name="ManufacturingAPI"):
    # 加载OpenAPI规范
    with open(api_schema_file, "r") as f:
        openapi_spec = json.load(f)
    
    # 创建Gateway目标
    api_target = self.client.create_mcp_gateway_target(
        gateway=gateway,
        name=target_name,
        target_type="openApiSchema",
        target_payload={"inlinePayload": json.dumps(openapi_spec)},
        credentials={
            "api_key": api_key,
            "credential_location": "QUERY_PARAMETER",  # 或 "HEADER"
            "credential_parameter_name": "api_key"
        }

def save_gateway_config(self, gateway, cognito_response, config_file="gateway_config.json"):
    config = {
        "gateway_url": gateway["gatewayUrl"],
        "gateway_id": gateway["gatewayId"],
        "region": self.region,
        "client_info": cognito_response["client_info"]
    }
    
    with open(f"gateway/{config_file}", "w") as f:
        json.dump(config, f, indent=2)

def setup_manufacturing_system():
    setup = ManufacturingGatewaySetup()
    
    # 1. 设置Gateway和认证
    gateway, cognito_response = setup.setup_manufacturing_gateway("ManufacturingSystem")
    
    # 2. 集成制造业API
    api_target = setup.add_manufacturing_api_target(
        gateway=gateway,
        api_schema_file="gateway/model_api_openapi_schema.json",
        api_key="your-manufacturing-api-key",
        target_name="ModelResourceAPI"
    )
    
    # 3. 保存配置
    config = setup.save_gateway_config(gateway, cognito_response)
    
    return config

gent运行的第一步是加载Gateway配置和初始化必要的客户端：
from bedrock_agentcore_starter_toolkit.operations.gateway.mcp_client import MCPClient
from bedrock_agentcore_starter_toolkit.operations.bedrock.client import BedrockClient
from bedrock_agentcore_starter_toolkit.operations.gateway.auth import get_access_token

class ManufacturingAgentRunner:
    def __init__(self, config_file="gateway/gateway_config.json"):
        self.config = self.load_gateway_config(config_file)
        self.gateway_url = self.config["gateway_url"]
        self.client_info = self.config["client_info"]
        self.region = self.config["region"]

async def setup_authentication(self):
    try:
        self.access_token = await get_access_token(
            client_id=self.client_info["client_id"],
            client_secret=self.client_info["client_secret"],
            cognito_domain=self.client_info["cognito_domain"]
        )
        return True
    except Exception as e:
        self.logger.error(f"认证失败: {e}")
        return False

def setup_bedrock_model(self, model_id="anthropic.claude-4-sonnet-20241022-v2:0"):
    try:
        self.bedrock_client = BedrockClient(region_name=self.region)
        bedrock_model = BedrockModel(
            model_id=model_id,
            client=self.bedrock_client
        )
        return bedrock_model
    except Exception as e:
        self.logger.error(f"Bedrock模型设置失败: {e}")
        return None

async def setup_mcp_client(self):
    try:
        self.mcp_client = MCPClient(
            gateway_url=self.gateway_url,
            access_token=self.access_token
        )
        await self.mcp_client.connect()
        return True
    except Exception as e:
        self.logger.error(f"MCP客户端连接失败: {e}")
        return False

async def get_available_tools(self) -> List[Dict[str, Any]]:
    try:
        tools = await self.mcp_client.list_tools()
        print(f"发现 {len(tools)} 个可用工具:")
        for tool in tools:
            print(f"  - {tool['name']}: {tool.get('description', '无描述')}")
        return tools
    except Exception as e:
        self.logger.error(f"获取工具列表失败: {e}")
        return []

async def run_manufacturing_query(self, query: str, bedrock_model) -> str:
    # 获取可用工具
    tools = await self.get_available_tools()
    
    # 构建制造业专用系统提示
    system_prompt = f"""
    你是一个制造业AI助手，专门帮助用户处理制造业相关的查询和任务。
    
    可用工具: {[tool["name"] for tool in tools]}
    
    请根据用户查询选择合适的工具并提供专业的制造业建议。
    """
    
    # 调用Bedrock模型
    response = await bedrock_model.ainvoke(
        input_text=query,
        system_prompt=system_prompt,
        max_tokens=2000,
        temperature=0.1  # 低温度确保准确性
    )
    
    return response

async def interactive_session(self):
    print("🏭 制造业AgentCore交互式会话")
    print("输入 'tools' 查看可用工具，'quit' 退出")
    
    bedrock_model = self.setup_bedrock_model()
    
    while True:
        user_input = input("🤖 请输入您的制造业查询: ").strip()
        
        if user_input.lower() in ['quit', 'exit']:
            break
        elif user_input.lower() == 'tools':
            await self.get_available_tools()
            continue
        
        response = await self.run_manufacturing_query(user_input, bedrock_model)
        print(f"📋 回答:\n{response}")

async def run_manufacturing_agent():
    runner = ManufacturingAgentRunner()
    
    try:
        # 1. OAuth认证
        await runner.setup_authentication()
        
        # 2. 连接Gateway
        await runner.setup_mcp_client()
        
        # 3. 发现工具
        await runner.get_available_tools()
        
        # 4. 启动交互会话
        await runner.interactive_session()
        
    finally:
        # 清理资源
        await runner.cleanup()

# 运行Agent
asyncio.run(run_manufacturing_agent())