用自然语言驱动数据看板：基于 Amazon Bedrock AgentCore 构建 Dashboard 自动化 Agent

pip install strands-agents boto3
# 如果要本地测试 QuickSight 调用，还需要配置 AWS CLI
aws configure set region us-east-1

from strands import Agent, Tool
import boto3
import json

# --- 工具定义 ---
@Tool(name="query_athena", description="在 Athena 中执行 SQL 查询，返回结果集")
def query_athena(sql: str, database: str = "sales_db") -> dict:
    """执行 Athena 查询，返回结果。生产环境应加结果大小限制和超时控制。"""
    athena = boto3.client("athena")
    # 启动查询
    response = athena.start_query_execution(
        QueryString=sql,
        QueryExecutionContext={"Database": database},
        ResultConfiguration={"OutputLocation": "s3://my-athena-results/"},
    )
    query_id = response["QueryExecutionId"]

    # 等待完成（简化版，生产环境应加轮询间隔和超时）
    import time
    while True:
        status = athena.get_query_execution(QueryExecutionId=query_id)
        state = status["QueryExecution"]["Status"]["State"]
        if state in ("SUCCEEDED", "FAILED", "CANCELLED"):
            break
        time.sleep(1)

    if state != "SUCCEEDED":
        return {"error": f"查询失败: {state}"}

    # 获取结果
    results = athena.get_query_results(QueryExecutionId=query_id)
    rows = []
    for row in results["ResultSet"]["Rows"][1:]:  # 跳过表头
        rows.append([col.get("VarCharValue", "") for col in row["Data"]])
    return {"columns": [col["VarCharValue"] for col in results["ResultSet"]["Rows"][0]["Data"]],
            "rows": rows}


@Tool(name="create_quicksight_dashboard",
      description="根据查询结果和用户意图，调用 QuickSight 创建或更新看板")
def create_quicksight_dashboard(intent: str, data: dict, dashboard_id: str = "auto-generated") -> dict:
    """将数据包装为 QuickSight 数据集并生成分析。
    实际生产中会调用 quicksight.create_data_set + create_analysis。
    这里展示核心调用逻辑。"""
    quicksight = boto3.client("quicksight", region_name="us-east-1")
    aws_account_id = boto3.client("sts").get_caller_identity()["Account"]

    # 根据意图选择可视化类型（简化逻辑）
    viz_map = {
        "趋势": "LINE_CHART",
        "对比": "BAR_CHART",
        "占比": "PIE_CHART",
        "分布": "SCATTER_CHART",
    }
    viz_type = "LINE_CHART"  # 默认
    for keyword, chart in viz_map.items():
        if keyword in intent:
            viz_type = chart
            break

    # 创建数据集（简化：用 Athena 结果 S3 位置作为源）
    dataset_response = quicksight.create_data_set(
        AwsAccountId=aws_account_id,
        DataSetId=f"ds-{dashboard_id}",
        Name=f"Agent 生成数据集 - {intent[:30]}",
        PhysicalTableMap={
            "athena-source": {
                "RelationalTable": {
                    "DataSourceArn": "arn:aws:quicksight:us-east-1:aws:datasource/Athena",
                    "Schema": "sales_db",
                    "Name": "regional_sales",
                    "InputColumns": [
                        {"Name": col, "Type": "STRING"} for col in data.get("columns", [])
                    ],
                }
            }
        },
        LogicalTableMap={
            "lt-1": {
                "Alias": "AgentQueryResult",
                "Source": {"PhysicalTableId": "athena-source"},
            }
        },
    )

    return {
        "status": "created",
        "dataset_id": dataset_response["DataSetId"],
        "viz_type": viz_type,
        "intent": intent,
    }


# --- Agent 定义 ---
dashboard_agent = Agent(
    name="dashboard-automation-agent",
    description="根据用户的自然语言需求，查询数据并生成 QuickSight 看板",
    tools=[query_athena, create_quicksight_dashboard],
    model="us.anthropic.claude-3-5-sonnet-20241022-v2:0",  # Bedrock 模型 ID
)

# --- 运行 ---
if __name__ == "__main__":
    result = dashboard_agent.invoke(
        "帮我看看华东区过去三个季度的退货率趋势，生成一个看板"
    )
    print(json.dumps(result, indent=2, ensure_ascii=False))

# agentcore-deployment.yaml
apiVersion: agentcore.aws/v1
kind: Agent
metadata:
  name: dashboard-automation-agent
  description: 自然语言驱动的数据看板生成与运维 Agent
spec:
  runtime:
    type: strands
    version: "0.5"
  model:
    provider: anthropic
    modelId: claude-3-5-sonnet
  tools:
    - name: query_athena
      type: custom
      iamRole: arn:aws:iam::123456789012:role/AgentCore-AthenaQueryRole
      config:
        database: sales_db
        outputLocation: s3://my-athena-results/
        resultSizeLimit: 1000
    - name: create_quicksight_dashboard
      type: custom
      iamRole: arn:aws:iam::123456789012:role/AgentCore-QuickSightRole
      config:
        allowedVizTypes: [LINE_CHART, BAR_CHART, PIE_CHART, SCATTER_CHART]
  security:
    sessionIsolation: true
    dataAccessScope:
      athenaWorkgroups: [primary]
      quickSightNamespace: default
    auditLog:
      cloudTrail: true
      s3Archive: s3://agentcore-audit-logs/

# 使用 AgentCore CLI 部署（假设已安装）
agentcore deploy --config agentcore-deployment.yaml --region us-east-1

# 验证 Agent 状态
agentcore describe-agent --name dashboard-automation-agent --region us-east-1