Claude Opus 4.8 登陆 Amazon Bedrock：智能体系统与生产推理的实战接入

pip install boto3 anthropic

import boto3
import json
import os

# ---- 配置 ----
REGION = "us-east-1"
MODEL_ID = "anthropic.claude-opus-4-8-20250514"  # 以 Bedrock 控制台实际 ID 为准

bedrock = boto3.client("bedrock-runtime", region_name=REGION)

# ---- 工具定义 ----
tools = [
    {
        "name": "read_file",
        "description": "读取指定路径的文件内容",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "文件路径"}
            },
            "required": ["path"]
        }
    },
    {
        "name": "search_issues",
        "description": "在代码中搜索潜在问题关键词",
        "input_schema": {
            "type": "object",
            "properties": {
                "keyword": {"type": "string", "description": "搜索关键词"},
                "directory": {"type": "string", "description": "搜索目录"}
            },
            "required": ["keyword", "directory"]
        }
    }
]

# ---- 工具执行器 ----
def execute_tool(name: str, input: dict) -> str:
    if name == "read_file":
        path = input["path"]
        if os.path.exists(path):
            with open(path, "r") as f:
                return f.read()[:4000]  # 限制长度避免上下文爆炸
        return f"文件不存在: {path}"
    elif name == "search_issues":
        keyword = input["keyword"]
        directory = input["directory"]
        results = []
        for root, _, files in os.walk(directory):
            for fname in files:
                if fname.endswith(".py"):
                    fpath = os.path.join(root, fname)
                    with open(fpath, "r") as f:
                        for i, line in enumerate(f, 1):
                            if keyword in line:
                                results.append(f"{fpath}:{i}: {line.strip()}")
        return "\n".join(results[:20]) if results else "未找到匹配"
    return "未知工具"

# ---- Agentic Loop ----
def run_agent(task: str, max_turns: int = 6) -> str:
    messages = [{"role": "user", "content": task}]
    system_prompt = "你是一个代码审查助手。先读取相关文件，再分析问题，最后给出具体修改建议。每一步都要调用工具获取信息，不要凭空猜测。"

    for _ in range(max_turns):
        # 构造 Bedrock 请求体
        body = {
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": 4096,
            "system": system_prompt,
            "messages": messages,
            "tools": [
                {
                    "name": t["name"],
                    "description": t["description"],
                    "input_schema": t["input_schema"]
                }
                for t in tools
            ]
        }

        response = bedrock.invoke_model(
            modelId=MODEL_ID,
            body=json.dumps(body),
            contentType="application/json",
            accept="application/json"
        )
        result = json.loads(response["body"].read())

        # 处理响应内容块
        assistant_content = result.get("content", [])
        messages.append({"role": "assistant", "content": assistant_content})

        # 检查是否有工具调用
        tool_calls = [b for b in assistant_content if b.get("type") == "tool_use"]
        if not tool_calls:
            # 模型给出了最终文本回复，结束循环
            final_text = "".join(
                b.get("text", "") for b in assistant_content if b.get("type") == "text"
            )
            return final_text

        # 执行工具并追加结果
        tool_results = []
        for tc in tool_calls:
            output = execute_tool(tc["name"], tc["input"])
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": tc["id"],
                "content": output
            })
        messages.append({"role": "user", "content": tool_results})

    return "达到最大轮次限制，智能体未完成任务。"

# ---- 运行 ----
if __name__ == "__main__":
    # 示例任务：审查当前目录下的 Python 代码
    task = "请审查 ./src 目录下的 Python 代码，重点关注潜在的异常处理缺失和硬编码密钥问题。"
    print(run_agent(task))

def route_model(task_complexity: str) -> str:
    """根据任务复杂度路由到不同模型"""
    routing = {
        "simple": "anthropic.claude-haiku-3-5-20241022",
        "moderate": "anthropic.claude-sonnet-4-20250514",
        "complex": "anthropic.claude-opus-4-8-20250514"
    }
    return routing.get(task_complexity, routing["moderate"])

# 通过 AWS CLI 查看当前预留吞吐配置
aws bedrock list-provisioned-model-throughputs --region us-east-1