技术分享 | 基于 Amazon Bedrock AgentCore + Web Search 构建 AI 竞争分析助手

2026-07-13 13:45

在快节奏的商业环境中，竞争情报是企业决策的重要依据。传统竞争分析依赖人工定期搜集信息、整理报告，不仅耗时且容易遗漏关键动态。

本文将展示如何利用亚马逊云科技最新发布的 Web Search on Amazon Bedrock AgentCore（2026 年 6 月 GA），构建一个能够实时联网搜索、自动分析并生成结构化竞争报告的 AI 助手。你只需要输入一句话，Agent 就自动执行多维度搜索、交叉验证、结构化分析，3 分钟内输出一份完整的竞争情报报告。

一、技术方案选型

为什么选择 Amazon Bedrock AgentCore + Web Search？

技术栈总览

二、整体架构图

Agent 接收到用户请求后，按照四个阶段依次执行：

Step 1--意图解析：确认目标竞品、时间范围、分析维度

Step 2--多维度并行搜索：生成 5 个精准的英文搜索查询，从产品、商业、市场、技术、生态五个维度采集信息

Step 3--交叉分析：信息去重合并、多源交叉验证、时间线排序、趋势识别

Step 4--报告生成：输出结构化竞争情报报告，包含来源引用和战略建议

三、动手实践：完整构建过程

前提条件：在 IDE 环境中安装好所有包和依赖

#环境要求

Python 3.11+

AWS CLI v2 (configured) boto3 >= 1.35.0

#安装依赖

pip install strands-agents strands-agents-tools boto3 mcp-proxy-for-aws

step 1 创建 AgentCore Gateway 并启用 Web Search工具

首先创建 AgentCore Gateway 并添加 Web Search 作为 Target：

import time

import boto3

gateway_client = boto3.client("bedrock-agentcore-control", region_name="us-east-1")

GATEWAY_ROLE_ARN = "arn:aws:iam::accountid:role/you-role-name"

#1. 创建 Gateway

gateway_response = gateway_client.create_gateway(

name="competitive-intel-gateway",

protocolType="MCP",

description="Gateway for Competitive Intelligence Agent",

roleArn=GATEWAY_ROLE_ARN,

authorizerType="AWS_IAM"

)

gateway_id = gateway_response["gatewayId"]

#2. 轮询等待 Gateway 状态变为 ACTIVE

while True:

status_response = gateway_client.get_gateway(gatewayIdentifier=gateway_id)

current_status = status_response["status"]

print(f"当前状态: {current_status}")

if current_status == "READY":

break

elif current_status in ["FAILED", "DELETING"]:

raise Exception(f"Gateway 创建失败，当前状态为: {current_status}")

time.sleep(5)

#3. 添加 Web Search Target

print("正在关联 Web Search 工具...")

gateway_client.create_gateway_target(

gatewayIdentifier=gateway_id,

name="web-search-tool",

targetConfiguration={

"mcp": {

"connector": {

"source": {"connectorId": "web-search"},

"configurations": [{"name": "WebSearch", "parameterValues": {}}],

}

credentialProviderConfigurations=[

{"credentialProviderType": "GATEWAY_IAM_ROLE"}å

)

print(f"Gateway {gateway_id} 已创建，Web Search 工具已成功启用！")

step 2 配置 Agentcore IAM 角色权限

创建 Runtime 执行角色的权限策略，并添加信任策略 (Trust Policy)，允许 bedrock-agentcore.amazonaws.com服务：

{

"Version": "2012-10-17",

"Statement": [

{

"Sid": "InvokeGateway",

"Effect": "Allow",

"Action": "bedrock-agentcore:InvokeGateway",

"Resource": "arn:aws:bedrock-agentcore:us-east-1:account-id:gateway/competitive-intel-gateway-1s9kpehlxd"

{

"Sid": "BedrockModelInvocation",

"Effect": "Allow",

"Action": [

"bedrock:InvokeModel",

"bedrock:InvokeModelWithRe-sponseStream"

"Resource": [

"arn:aws:bedrock:*::foundation-model/*",

"arn:aws:bedrock:us-east-1:account-id:*"

]

{

"Sid": "InvokeWebSearch",

"Effect": "Allow",

"Action": "bedrock-agentcore:InvokeWebSearch",

"Resource": "arn:aws:bedrock-agentcore:us-east-1:aws:tool/web-search.v1"

}

]

}

step 3 构建竞争分析 Agent

这是核心代码，包含完整的竞争分析工作流。Agent 通过 System Prompt 定义分析方法论，通过 MCP 协议自动发现并调用 Web Search 工具：

"""

competitive_intel_agent.py

竞争分析 AI 助手 - 基于 Amazon Bedrock AgentCore + Web Search

"""

from datetime import date

from strands import Agent

from strands.models.bedrock import BedrockModel

from strands.tools.mcp import MCPClient

from mcp_proxy_for_aws.client import aws_iam_streamablehttp_client

#============================================================

#配置

#============================================================

GATEWAY_URL = "https://competitive-intel-gateway-1s9kpehlxd.gateway.bedrock-agentcore.us-east-1.amazonaws.com/mcp" #替换成自己账号中的url

AWS_REGION = "us-east-1"

MODEL_ID = "xxxxx"

#============================================================

#竞争分析专用 System Prompt

#============================================================

##你的角色

你为产品团队和战略团队提供实时、准确、有深度的竞品分析报告。你擅长从公开信息中洞察竞品的战略意图和市场动向

##分析方法论

当用户提出竞品分析需求时，严格按照以下流程执行：

###Phase 1: 需求理解

- 确认目标竞品名称

- 确认分析时间范围

- 确认关注的分析维度（产品/定价/市场/技术/生态）

###Phase 2: 多维度信息采集

对每个分析维度，制定 2-3 个精准的英文搜索查询（英文搜索覆盖面更广），从不同角度收集信息：

- **产品动态**: 新功能发布、产品路线图、版本更新

- **商业策略**: 定价变化、目标市场转移、营销活动

- **市场表现**: 分析师评价、市场份额、客户反馈

- **技术架构**: 技术博客、开源贡献、技术选型变化

- **生态合作**: 新合作伙伴、集成、收购

###Phase 3: 分析与洞察

- 识别关键事件并排列时间线

- 发现趋势和模式

- 评估对我方的潜在影响

- 提出应对建议

###Phase 4: 输出报告

严格按照以下格式输出：

---

#竞争情报报告: [竞品名称]

**分析时间范围**: [时间范围]

**报告生成时间**: {date.today().isoformat()}

**信息来源数**: [N] 条

#核心发现（Executive Summary）

> [3-5 句话概括最重要的发现]

##关键事件时间线

| 日期 | 事件 | 影响程度 | 来源 |

|------|------|---------|------|

| ... | ... | 高/中/低 | [链接] |

##产品动态分析

###新功能与更新

[细节分析]

###技术方向判断

[基于产品动态推断的技术路线]

##商业策略分析

###定价与包装变化

[如有变化，详细说明]

###目标市场与定位

[市场动向分析]

##生态与合作

[合作伙伴、集成、收购等动态]

##市场声音

###分析师观点

[来自 Gartner、Forrester 等的评价]

###客户与社区反馈

[开发者社区、社交媒体的讨论]

##对我方的影响评估

| 影响维度 | 影响程度 | 说明 |

|---------|---------|------|

| ... | 高/中/低 | ... |

##应对建议

1. [短期行动建议]

2. [中期策略调整]

3. [长期需要关注的方向]

##信息来源

[编号列出所有来源的标题、URL、发布时间]

##后续追踪建议

[建议持续关注的信号和触发条件]

---

##重要原则

- 每条重要信息必须标注来源 URL

- 区分"事实"和"推测"，推测部分需明确标注

- 如果某个维度没有搜索到有效信息，诚实说明

- 英文搜索优先（覆盖面更广），中文补充搜索

- 关注最近 3 个月的信息，优先呈现最新动态

"""

#============================================================

#Agent 创建与执行

#============================================================

def create_competitive_intel_agent():

mcp_client = MCPClient(lambda: aws_iam_streamablehttp_client(

endpoint=GATEWAY_URL,

aws_region=AWS_REGION,

aws_service="bedrock-agentcore",

))

model = BedrockModel(model_id=MODEL_ID)

return mcp_client, model

def analyze_competitor(query: str) -> str:

"""

执行竞争分析

Args:

query: 用户的分析需求，例如 "分析 Datadog 近三个月的产品动态"

Returns:

结构化的竞争情报报告

"""

mcp_client, model = create_competitive_intel_agent()

with mcp_client:

#自动发现 Gateway 上的 Web Search 工具

tools = mcp_client.list_tools_sync()

print(f"已连接 AgentCore Gateway, 发现 {len(tools)} 个工具")

#创建 Agent

agent = Agent(

model=model,

tools=tools,

system_prompt=SYSTEM_PROMPT

)

#执行分析

print(f"开始竞争分析: {query}")

print("=" * 60)

result = agent(query)

print("=" * 60)

print("分析完成")

return result

class CompetitiveIntelTracker:

"""持续追踪模式 - 记住历史分析，识别变化趋势"""

def __init__(self, memory_store_id: str):

self.memory_store_id = memory_store_id

self.mcp_client, self.model = create_competitive_intel_agent()

def analyze(self, query: str, session_id: str = "default") -> str:

"""带上下文记忆的分析"""

with self.mcp_client:

tools = self.mcp_client.list_tools_sync()

agent = Agent(

model=self.model,

tools=tools,

system_prompt=SYSTEM_PROMPT + f"""

\n\n##历史追踪模式

你正在持续追踪模式下工作。利用 Memory 中的历史分析记录：

- 对比本次发现与上次分析的变化

- 标注"新发现"vs"持续趋势"

- 在报告末尾增加"与上次分析对比"章节

""",

memory_config={

"memory_store_id": self.memory_store_id,

"session_id": session_id,

"namespace": "competitive-intel"

}

)

return agent(query)

step 4 打包为容器镜像并推送到 ECR

Step 3的代码可以直接在本地运行（用于开发调试），但要部署到 AgentCore Runtime 作为生产服务，需要将其打包为 Docker arm64 架构镜像并推送到 Amazon ECR。

4.1 创建 Dockerfile

FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY competitive_intel_agent.py . COPY entrypoint.py . EXPOSE 8080 CMD ["python", "entrypoint.py"]

4.2 创建 entrypoint.py（HTTP 服务入口）

AgentCore Runtime 通过 HTTP 调用 Agent 容器，需要一个服务入口：

"""entrypoint.py - AgentCore Runtime 入口"""

from fastapi import FastAPI, Request

from competitive_intel_agent import analyze_competitor

import uvicorn

import traceback

app = FastAPI()

@app.post("/invoke")

@app.post("/invocations")

@app.post("/")

async def invoke(request: Request):

try:

body = await request.json()

query = body.get("prompt") or body.get("query") or body.get("input", "")

session_id = body.get("session_id", "default")

if not query:

return {"error": "No prompt/query provided", "received_body": body}

result = analyze_competitor(query)

return {"response": str(result)}

except Exception as e:

traceback.print_exc()

return {"error": str(e), "traceback": traceback.format_exc()}

#健康检查（AgentCore 可能用来探活）

@app.get("/health")

@app.get("/ping")

@app.get("/")

async def health():

return {"status": "healthy"}

if __name__ == "__main__":

uvicorn.run(app, host="0.0.0.0", port=8080)

4.3 创建 requirements.txt

strands-agents>=0.3.0 strands-agents-tools>=0.3.0 boto3>=1.35.0 mcp-proxy-for-aws>=0.1.0 fastapi>=0.100.0 uvicorn>=0.23.0

4.4

构建并推送到 ECR

#创建 ECR 仓库 aws ecr create-repository --repository-name competitive-intel-agent --region us-east-1 #登录 ECR aws ecr get-login-password --region us-east-1 | \ docker login --username AWS --password-stdin <ACCOUNT_ID>.dkr.ecr.us-east-1.amazonaws.com #构建镜像 docker build -t competitive-intel-agent . #打 Tag 并推送 docker tag competitive-intel-agent:latest \ <ACCOUNT_ID>.dkr.ecr.us-east-1.amazonaws.com/competitive-intel-agent:latest docker push <ACCOUNT_ID>.dkr.ecr.us-east-1.amazonaws.com/competitive-intel-agent:latest