AgentOps：让 AI Agent 在生产环境真正可控

pip install boto3
aws configure  # 填入 Access Key、Secret Key、region（如 us-east-1）

import boto3
import time
import json

client = boto3.client("bedrock-agent", region_name="us-east-1")

# 1. 创建 Agent
create_resp = client.create_agent(
    agentName="ops-demo-agent",
    agentResourceRoleArn="arn:aws:iam::<YOUR_ACCOUNT_ID>:role/BedrockAgentRole",  # 替换为你的 IAM Role
    foundationModel="anthropic.claude-3-5-sonnet-20241022-v2:0",
    instruction="你是一个运维助手。用户报告服务异常时，先查询日志，再给出诊断建议。不要猜测，必须基于工具返回的数据回答。",
    idleSessionTTLInSeconds=600,  # 会话超时 10 分钟
)

agent_id = create_resp["agent"]["agentId"]
print(f"Agent ID: {agent_id}")

# 等待 Agent 创建完成
time.sleep(10)

# 2. 创建 Guardrail —— 限制输出内容 + 限制工具调用范围
guard_resp = client.create_guardrail(
    name="ops-demo-guardrail",
    description="限制 Agent 只讨论运维话题，拒绝无关请求",
    contentPolicyConfig={
        "filtersConfig": [
            {
                "type": "DENIED",
                "inputStrength": "HIGH",
                "outputStrength": "HIGH",
                "topics": ["金融投资", "医疗诊断"]  # 拒绝这些话题
            }
        ]
    },
    contextualGroundingPolicyConfig={
        "filtersConfig": [
            {
                "type": "GROUNDING",
                "threshold": 0.75  # 要求输出至少 75% 有依据
            },
            {
                "type": "RELEVANCE",
                "threshold": 0.80  # 要求输出与问题相关度 ≥ 80%
            }
        ]
    },
    blockedInputMessaging="抱歉，我只能协助处理运维相关的问题。",
    blockedOutputsMessaging="输出被护栏拦截：内容缺乏充分依据或偏离运维领域。",
)

guardrail_id = guard_resp["guardrailId"]
print(f"Guardrail ID: {guardrail_id}")

# 3. 将 Guardrail 绑定到 Agent
client.update_agent(
    agentId=agent_id,
    guardrailConfiguration={
        "guardrailId": guardrail_id,
        "guardrailVersion": "DRAFT",
    },
)

print("Agent 已绑定护栏，准备进入 DRAFT 版本测试。")

# 4. 创建一个模拟的日志查询工具（Lambda 函数）
lambda_client = boto3.client("lambda", region_name="us-east-1")

lambda_client.create_function(
    FunctionName="ops-log-query-tool",
    Runtime="python3.12",
    Role="arn:aws:iam::<YOUR_ACCOUNT_ID>:role/LambdaExecutionRole",  # 替换
    Handler="index.handler",
    Code={"ZipFile": b"""
import json
def handler(event, context):
    # 模拟日志查询结果
    service = event.get("service", "unknown")
    return {
        "statusCode": 200,
        "body": json.dumps({
            "service": service,
            "last_errors": [
                {"time": "2025-01-15T08:32:00Z", "message": "Connection timeout to DB"},
                {"time": "2025-01-15T08:33:15Z", "message": "Retry succeeded"},
            ],
            "error_rate_last_1h": "0.03%"
        })
    }
"""},
)

time.sleep(5)

# 5. 注册工具到 Agent
tool_resp = client.create_agent_action_group(
    agentId=agent_id,
    agentVersion="DRAFT",
    actionGroupName="LogQuery",
    actionGroupExecutor={
        "lambda": "arn:aws:lambda:us-east-1:<YOUR_ACCOUNT_ID>:function:ops-log-query-tool"
    },
    functionSchema={
        "functions": [
            {
                "name": "query_service_logs",
                "description": "查询指定服务的最近错误日志和错误率",
                "parameters": {
                    "service": {
                        "type": "string",
                        "description": "服务名称，如 order-service、payment-service",
                        "required": True,
                    }
                },
            }
        ]
    },
)

print(f"工具已注册: {tool_resp['actionGroup']['actionGroupId']}")

# 6. 准备 Agent 版本（DRAFT → 可调用）
prepare_resp = client.prepare_agent(agentId=agent_id)
print(f"Agent 版本已准备: {prepare_resp['preparedAgent']['agentVersion']}")

runtime = boto3.client("bedrock-agent-runtime", region_name="us-east-1")

# 7. 发起一次对话，启用轨迹记录
resp = runtime.invoke_agent(
    agentId=agent_id,
    agentAliasId="TSTALIASID",  # 测试别名
    sessionId="ops-session-001",
    inputText="order-service 最近有异常吗？帮我查一下日志。",
    enableTrace=True,  # 关键：开启轨迹采集
)

# 8. 解析轨迹数据
trace_data = []
for event in resp.get("completion", []):
    if "trace" in event:
        trace = event["trace"]
        trace_data.append(trace)
        # 打印每一步决策
        if "orchestrationTrace" in trace:
            ort = trace["orchestrationTrace"]
            if "modelInvocationOutput" in ort:
                reasoning = ort["modelInvocationOutput"].get("rawResponse", "")
                print(f"[推理] {reasoning[:200]}...")
            if "actionInvocationOutput" in ort:
                action = ort["actionInvocationOutput"]
                print(f"[工具调用] function={action.get('functionName')} input={action.get('actionInput')}")

    if "chunk" in event:
        answer = event["chunk"].get("bytes", b"").decode("utf-8")
        print(f"\n[最终回答] {answer}")

# 9. 将轨迹保存为 JSON，供后续分析
with open("agent_trace_session_001.json", "w") as f:
    json.dump(trace_data, f, indent=2, ensure_ascii=False)

print(f"\n轨迹已保存，共 {len(trace_data)} 个决策节点。")

# 10. 创建带执行限制的 Agent Alias
alias_resp = client.create_agent_alias(
    agentAliasName="prod-alias",
    agentId=agent_id,
    routingConfiguration=[
        {
            "agentVersion": "1",  # 使用已发布的版本
        }
    ],
    # Bedrock AgentCore 支持在会话级别设置 maxSteps 等参数
    # 通过 session-level 配置传入：
    #   maxSteps: 5        # 单次任务最多 5 步
    #   maxTokensPerStep: 4096
)

print(f"生产别名已创建: {alias_resp['agentAlias']['agentAliasId']}")

resp = runtime.invoke_agent(
    agentId=agent_id,
    agentAliasId=alias_resp["agentAlias"]["agentAliasId"],
    sessionId="ops-session-002",
    inputText="payment-service 响应变慢了，帮我排查。",
    enableTrace=True,
    sessionState={
        "invocationId": "inv-002",
        # 限制单次对话最多 5 轮工具调用
        # （具体参数名以 Bedrock AgentCore 最新 SDK 文档为准）
    },
)