AI代理从概念验证到生产部署全流程

from strands import Agent
from strands.models import BedrockModel

@tool
def get_return_policy(product_category: str) -> str:
    """获取特定产品类别的退货政策信息。"""
    # 返回结构化政策信息：窗口、条件、流程、退款
    # 查看github获取完整代码
    return {"return_window": "10天", "conditions": ""}
    
@tool  
def get_product_info(product_type: str) -> str:
    """获取电子产品的详细技术规格和信息。"""
    # 返回保修、规格、功能、兼容性详情
    # 查看github获取完整代码
    return {"product": "ThinkPad X1 Carbon", "info": "ThinkPad X1 Carbon信息"}
    
@tool
def web_search(keywords: str, region: str = "us-en", max_results: int = 5) -> str:
    """搜索网络以获取更新的故障排除信息。"""
    # 提供对当前技术解决方案和指南的访问
    # 查看github获取完整代码
    return "网络搜索结果"
    
# 初始化模型
model = BedrockModel(
    model_id="us.anthropic.claude-3-7-sonnet-20250219-v1:0",
    temperature=0.3
)

# 创建客户支持代理
agent = Agent(
    model=model,
    tools=[
        get_product_info, 
        get_return_policy, 
        web_search
    ],
    system_prompt="""您是一家电子产品公司的有用客户支持助手。
    使用适当的工具提供准确信息，并始终提供额外帮助。"""
)

# 退货政策查询
response = agent("我的ThinkPad X1 Carbon的退货政策是什么？")
# 代理正确使用get_return_policy与"笔记本电脑"类别

# 技术故障排除
response = agent("我的iPhone 14发热，如何修复？")
# 代理使用web_search查找当前故障排除解决方案

pip install boto3 bedrock-agentcore bedrock-agentcore-starter-toolkit

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

memory_client = MemoryClient(region_name=region)

strategies = [
    {
        StrategyType.USER_PREFERENCE.value: {
            "name": "CustomerPreferences",
            "description": "捕获客户偏好和行为",
            "namespaces": ["support/customer/{actorId}/preferences"],
        }
    },
    {
        StrategyType.SEMANTIC.value: {
            "name": "CustomerSupportSemantic", 
            "description": "存储对话中的事实",
            "namespaces": ["support/customer/{actorId}/semantic"],
        }
    },
]

# 使用两种策略创建内存资源
response = memory_client.create_memory_and_wait(
    name="CustomerSupportMemory",
    description="客户支持代理内存",
    strategies=strategies,
    event_expiry_days=90,
)

class CustomerSupportMemoryHooks(HookProvider):
    def retrieve_customer_context(self, event: MessageAddedEvent):
        """在处理查询前注入客户上下文"""
        user_query = event.agent.messages[-1]["content"][0]["text"]
        
        # 从两种策略检索相关记忆
        all_context = []
        for context_type, namespace in self.namespaces.items():
            memories = self.client.retrieve_memories(
                memory_id=self.memory_id,
                namespace=namespace.format(actorId=self.actor_id),
                query=user_query,
                top_k=3,
            )
            # 格式化并添加到上下文
            for memory in memories:
                if memory.get("content", {}).get("text"):
                    all_context.append(f"[{context_type.upper()}] {memory['content']['text']}")
        
        # 将上下文注入用户查询
        if all_context:
            context_text = "\n".join(all_context)
            original_text = event.agent.messages[-1]["content"][0]["text"]
            event.agent.messages[-1]["content"][0]["text"] = f"客户上下文:\n{context_text}\n\n{original_text}"

    def save_support_interaction(self, event: AfterInvocationEvent):
        """在代理响应后保存交互"""
        # 获取最后客户查询和代理响应 查看github获取实现
        customer_query = "这是一个示例查询"
        agent_response = "LLM给出了示例响应"
        
        # 提取客户查询和代理响应
        # 保存到内存以供将来检索
        self.client.create_event(
            memory_id=self.memory_id,
            actor_id=self.actor_id,
            session_id=self.session_id,
            messages=[(customer_query, "USER"), (agent_response, "ASSISTANT")]
        )

# 为此客户会话创建内存钩子
memory_hooks = CustomerSupportMemoryHooks(
    memory_id=memory_id, 
    client=memory_client, 
    actor_id=customer_id, 
    session_id=session_id
)

# 创建具有内存能力的代理
agent = Agent(
    model=model,
    tools=[get_product_info, get_return_policy, web_search],
    system_prompt=SYSTEM_PROMPT
)

# 测试个性化推荐
response = agent("您推荐哪款耳机？")
# 代理记住："偏好竞技FPS游戏的低延迟"
# 响应包括以游戏为重点的推荐

# 测试偏好回忆
response = agent("我偏好的笔记本电脑品牌是什么？")  
# 代理记住："偏好ThinkPad型号"和"需要Linux兼容性"
# 响应确认ThinkPad偏好并建议兼容型号

# 原始Lambda函数
def web_search(keywords: str, region: str = "us-en", max_results: int = 5) -> str:
    # web_search功能
        
def lambda_handler(event, context):
    if get_tool_name(event) == "web_search":
        query = get_named_parameter(event=event, name="query")
        
        search_result = web_search(keywords)
        return {"statusCode": 200, "body": search_result}

{
    "name": "web_search",
    "description": "使用DuckDuckGo搜索网络以获取更新信息",
    "inputSchema": {
        "type": "object",
        "properties": {
            "keywords": {
                "type": "string",
                "description": "搜索查询关键词"
            },
            "region": {
                "type": "string",
                "description": "搜索区域"
            },
            "max_results": {
                "type": "integer",
                "description": "返回的最大结果数"
            }
        },
        "required": [
            "keywords"
        ]
    }
}

# 创建基于JWT身份验证的网关
auth_config = {
    "customJWTAuthorizer": {
        "allowedClients": [cognito_client_id],
        "discoveryUrl": cognito_discovery_url
    }
}

gateway_response = gateway_client.create_gateway(
    name="customersupport-gw",
    roleArn=gateway_iam_role,
    protocolType="MCP",
    authorizerType="CUSTOM_JWT",
    authorizerConfiguration=auth_config,
    description="客户支持AgentCore网关"
)

lambda_target_config = {
    "mcp": {
        "lambda": {
            "lambdaArn": lambda_function_arn,
            "toolSchema": {"inlinePayload": api_spec},
        }
    }
}

gateway_client.create_gateway_target(
    gatewayIdentifier=gateway_id,
    name="LambdaTools",
    targetConfiguration=lambda_target_config,
    credentialProviderConfigurations=[{
        "credentialProviderType": "GATEWAY_IAM_ROLE"
    }]
)

# 获取网关访问的OAuth令牌
gateway_access_token = get_token(
    client_id=cognito_client_id,
    client_secret=cognito_client_secret,
    scope=auth_scope,
    url=token_url
)

# 创建经过身份验证的MCP客户端
mcp_client = MCPClient(
    lambda: streamablehttp_client(
        gateway_url,
        headers={"Authorization": f"Bearer {gateway_access_token['access_token']}"}
    )
)

# 组合本地和MCP工具
tools = [
    get_product_info,     # 本地工具
    get_return_policy,    # 本地工具
] + mcp_client.list_tools_sync()  # 来自网关的集中式工具

agent = Agent(
    model=model,
    tools=tools,
    hooks=[memory_hooks],
    system_prompt=SYSTEM_PROMPT
)

# 使用集中式工具测试网络搜索
response = agent("如何修复Lenovo ThinkPad蓝屏问题？")
# 代理使用来自AgentCore Gateway的web_search

# 您现有的代理代码保持不变
model = BedrockModel(model_id="us.anthropic.claude-3-7-sonnet-20250219-v1:0")
memory_hooks = CustomerSupportMemoryHooks(memory_id, memory_client, actor_id, session_id)
agent = Agent(
    model=model,
    tools=[get_return_policy, get_product_info],
    system_prompt=SYSTEM_PROMPT,
    hooks=[memory_hooks]
)

def invoke(payload):
    user_input = payload.get("prompt", "")
    response = agent(user_input)
    return response.message["content"][0]["text"]

if __name__ == "__main__":

from bedrock_agentcore_starter_toolkit import Runtime

# 使用身份验证配置安全部署
agentcore_runtime = Runtime()

response = agentcore_runtime.configure(
    entrypoint="lab_helpers/lab4_runtime.py",
    execution_role=execution_role_arn,
    auto_create_ecr=True,
    requirements_file="requirements.txt",
    region=region,
    agent_name="customer_support_agent",
    authorizer_configuration={
        "customJWTAuthorizer": {
            "allowedClients": [cognito_client_id],
            "discoveryUrl": cognito_discovery_url,
        }
    }
)

# 部署到生产环境
launch_result = agentcore_runtime.launch()

# 客户1对话
response1 = agentcore_runtime.invoke(
    {"prompt": "我的iPhone蓝牙不工作。我该怎么办？"},
    bearer_token=auth_token,
    session_id="session-customer-1"
)

# 客户1跟进
response2 = agentcore_runtime.invoke(
    {"prompt": "我已经打开和关闭蓝牙但仍然不工作"},
    bearer_token=auth_token,
    session_id="session-customer-1"  # 相同会话，上下文保留
)

# 客户2对话
response3 = agentcore_runtime.invoke(
    {"prompt": "仍然不工作。怎么回事？"},
    bearer_token=auth_token,
    session_id="session-customer-2"  # 不同会话，无上下文
)