基于Amazon Bedrock Agents构建多智能体协作AI团队实战指南

🚀 30+款热门AI模型一站整合，DeepSeek/GLM/Qwen 随心用，限时 5 折。 👉 点击领海量免费额度

在实际企业级AI应用开发中，构建一个能理解复杂意图、调用工具、处理多步骤任务的智能体（Agent）是核心挑战。传统的单一大模型调用往往只能完成一次性的问答或生成，而真正的业务自动化，如订单处理、客户服务、数据分析，需要AI能够自主规划、决策并执行一系列动作。Amazon Bedrock Agents 正是为解决这类“多步骤任务自动化”而设计的服务，它让开发者能够基于强大的基础模型（FM），快速构建、部署和管理具备推理、记忆和工具调用能力的AI智能体。

本文将以一个贴近实战的“代码秀”场景为例，带你从零开始，完整拆解如何使用 Amazon Bedrock Agents 构建一个能在技术峰会上进行现场代码演示的AI团队。这个团队将由多个分工明确的Agent组成，例如一个负责理解用户需求的“需求分析师Agent”，一个负责生成代码片段的“程序员Agent”，以及一个负责解释代码逻辑的“讲解员Agent”。我们将深入探讨其核心概念、配置步骤、代码实现、多Agent协作机制，并重点分析在实际部署中可能遇到的配置、权限、推理逻辑等问题及其排查路径。无论你是希望将AI能力集成到现有系统的开发者，还是对Agentic AI（智能体AI）架构感兴趣的技术决策者，本文都将提供一套可落地、可复现的工程实践指南。

1. 理解 Amazon Bedrock Agents 的核心工作机制

在开始动手之前，必须厘清几个关键概念，这决定了你能否正确设计和使用Agent。

1.1 Agent 是什么？不仅仅是聊天机器人

一个AI Agent（智能体）是一个能够感知环境、进行推理、制定计划并执行行动以实现特定目标的软件实体。在Amazon Bedrock的语境下，一个Agent由以下几个核心部分组成：

1. 基础模型（Foundation Model）：Agent的“大脑”，负责理解用户输入的自然语言指令，并进行推理和规划。Bedrock支持多种模型，如Anthropic Claude、Meta Llama等。
2. 指令（Instructions）：用自然语言描述Agent的角色、职责和行为边界。例如，“你是一个专业的Python后端开发专家，专注于生成高效、可读的代码。你会先分析需求，再给出代码示例和简要解释。”
3. 行动组（Action Groups）：定义了Agent可以执行的具体操作。每个行动组对应一个或多个API调用（通过OpenAPI Schema定义）。例如，一个“代码执行”行动组可能包含“运行Python代码片段”和“获取执行结果”两个API。
4. 知识库（Knowledge Bases）：可选组件。Agent可以连接到企业数据源（如S3、数据库），通过检索增强生成（RAG）技术，在回答问题时引用最新的、特定的公司知识，而不是仅依赖模型的通用知识。
5. 记忆（Memory）：Agent能够保留跨对话轮次的历史信息，从而实现连续、连贯的多步骤任务处理。

与简单的聊天机器人不同，一个配置了行动组的Bedrock Agent，在收到用户请求后，其内部工作流程是：理解 -> 规划 -> 执行 -> 汇总。它会自动判断是否需要调用工具（API）、需要调用哪个工具、传递什么参数，并在获得工具执行结果后，综合所有信息生成最终回复。

1.2 多Agent协作：构建一个“AI团队”

单个Agent的能力是有限的。复杂的业务场景，如我们设想的“代码秀”，需要多个Agent协同工作。这就是Bedrock的多Agent协作功能。

• 监督员Agent（Supervisor Agent）：负责接收用户的总任务，并将其分解为子任务，然后分配给最合适的专业Agent。
• 专业Agent（Specialist Agent）：每个专业Agent负责一个特定领域。在我们的场景中：
  • 需求分析Agent：擅长将模糊的用户描述（如“帮我写一个快速排序”）转化为结构化的技术需求文档。
  • 代码生成Agent：精通特定编程语言（如Python、Java），根据结构化需求生成代码。
  • 代码审查/解释Agent：负责分析生成的代码，解释其逻辑、复杂度，或指出潜在问题。

监督员Agent协调整个流程，决定任务流转顺序，并汇总各专业Agent的输出，最终呈现给用户一个完整的结果。这种架构模拟了真实世界的团队协作，极大地提升了处理复杂任务的可靠性和质量。

1.3 为什么选择 Bedrock Agents？关键特性与工程价值

2. 环境准备与项目初始化

在开始构建“代码秀”AI团队之前，需要完成基础环境搭建。这里假设你已拥有一个AWS账户并具备基本的IAM操作权限。

2.1 AWS 环境与权限配置

1. 启用 Bedrock 服务：首次使用前，需要在AWS管理控制台搜索并进入“Amazon Bedrock”服务，在概览页点击“启用模型访问”。根据需求选择需要使用的模型（例如Claude 3 Sonnet）。
2. 创建IAM策略与角色：这是最关键也最容易出错的一步。Agent要调用Lambda函数或外部API，需要一个具有相应权限的执行角色。
   • 策略示例：创建一个名为BedrockAgentExecutionPolicy的内联策略。

     { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "bedrock:InvokeModel", "bedrock:ListFoundationModels" ], "Resource": "*" }, { "Effect": "Allow", "Action": [ "lambda:InvokeFunction" ], "Resource": "arn:aws:lambda:<region>:<account-id>:function:code-show-*" }, { "Effect": "Allow", "Action": [ "s3:GetObject" ], "Resource": "arn:aws:s3:::code-show-knowledge-base/*" } ] }

   • 创建角色：在IAM控制台创建角色，信任实体选择“Bedrock”。将上述策略附加到该角色。记下角色的ARN，后续创建Agent时需要。

2.2 定义“代码秀”的业务流程与Agent职责

在编码之前，必须明确设计。我们的“代码秀”场景流程如下：

1. 用户提出请求：“用Python演示一下快速排序算法，并比较其与冒泡排序的性能。”
2. 监督员Agent接收请求，识别出其中包含多个子任务：a) 生成快速排序代码，b) 生成冒泡排序代码，c) 性能比较。
3. 监督员将任务a和b分配给代码生成Agent，将任务c分配给代码分析Agent。
4. 代码生成Agent调用一个“代码执行Lambda函数”，实际运行生成的排序代码，获取执行时间等数据。
5. 代码分析Agent接收两份代码和性能数据，生成一份对比分析报告。
6. 监督员Agent汇总代码片段、执行结果和分析报告，形成最终回复给用户。

基于此，我们需要创建三个Agent，并为其准备两个后端工具（Lambda函数）。

2.3 创建后端工具（Lambda函数）

Agent通过API调用工具。我们将创建两个简单的Lambda函数作为工具。

工具1：code-executor-python(Python代码执行器)

• 功能：接收一段Python代码，在一个安全的沙箱环境（例如使用subprocess）中执行，并返回输出或错误信息。
• Lambda代码示例 (Python 3.12)：

  import json import subprocess import sys import tempfile import os def lambda_handler(event, context): # 从Agent的请求中解析代码 # Agent调用时，参数会放在 `event['parameters']` 列表中 code_to_run = None for param in event.get('parameters', []): if param.get('name') == 'code': code_to_run = param.get('value') break if not code_to_run: return { 'statusCode': 400, 'body': json.dumps({'error': 'No code provided in parameters'}) } # 使用临时文件和安全配置执行代码 with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f: f.write(code_to_run) temp_file_path = f.name try: # 限制执行时间和资源 result = subprocess.run( [sys.executable, temp_file_path], capture_output=True, text=True, timeout=10, # 超时10秒 cwd=tempfile.gettempdir() ) output = { 'stdout': result.stdout, 'stderr': result.stderr, 'returncode': result.returncode } return { 'statusCode': 200, 'body': json.dumps(output) } except subprocess.TimeoutExpired: return { 'statusCode': 408, 'body': json.dumps({'error': 'Code execution timeout'}) } except Exception as e: return { 'statusCode': 500, 'body': json.dumps({'error': f'Execution failed: {str(e)}'}) } finally: # 清理临时文件 try: os.unlink(temp_file_path) except: pass

• 注意：生产环境需要更严格的沙箱隔离，考虑使用容器或专用安全服务。

工具2：code-analyzer(代码分析器)

• 功能：接收两段代码及其运行性能数据，生成简单的文本对比报告。
• Lambda代码示例：

  import json def lambda_handler(event, context): # 解析参数 code_a = None code_b = None perf_a = None perf_b = None for param in event.get('parameters', []): name = param.get('name') value = param.get('value') if name == 'code_a': code_a = value elif name == 'code_b': code_b = value elif name == 'perf_a': perf_a = value elif name == 'perf_b': perf_b = value # 简单的分析逻辑（实际可集成更复杂的静态分析工具） analysis = f""" 代码分析报告： 1. 代码A (快速排序)： - 行数：{len(code_a.splitlines())} - 时间复杂度：平均 O(n log n)，最坏 O(n^2) - 空间复杂度：O(log n) - 实测执行时间：{perf_a if perf_a else 'N/A'} 2. 代码B (冒泡排序)： - 行数：{len(code_b.splitlines())} - 时间复杂度：O(n^2) - 空间复杂度：O(1) - 实测执行时间：{perf_b if perf_b else 'N/A'} 3. 对比结论： - 对于大规模数据，快速排序通常远快于冒泡排序。 - 冒泡排序实现简单，但效率低，仅适用于极小数据集教学。 """ return { 'statusCode': 200, 'body': json.dumps({'analysis': analysis}) }

创建Lambda后，记下它们的ARN。同时，需要为每个Lambda创建一个别名（Alias），如PROD，Agent将调用该别名，便于未来无缝更新函数版本。

3. 构建“代码秀”AI团队：配置与集成

我们将通过AWS管理控制台（或AWS CLI/SDK）逐步配置三个Agent。控制台操作直观，适合首次理解流程。

3.1 配置代码生成 Agent

1. 进入Bedrock控制台：导航至“Agents”部分，点击“Create agent”。
2. 基础设置：
   • Agent name:CodeGeneratorAgent
   • Description: “专业Python代码生成与执行Agent。”
   • IAM role: 选择在2.1步骤中创建的角色。
   • Model: 选择Claude 3 Sonnet（平衡性能与成本）。
3. 指令（Instructions）：这是Agent的“人格”设定，至关重要。

   你是一个专业的Python开发专家。你的职责是根据用户需求生成正确、高效、可读的Python代码。 如果用户的需求描述不够具体，你应该主动询问澄清，例如询问输入数据的格式、函数的签名、或性能要求。 你拥有一个可以执行Python代码的工具。在生成代码后，如果条件允许且用户未明确禁止，你应该主动调用该工具来验证代码是否能正常运行，并将执行结果一并返回。 你的回复应包含：1) 生成的代码（用代码块包裹），2) 对代码逻辑的简要说明，3) 代码的执行结果（如果执行了的话）。 不要生成有害代码或执行危险操作。

4. 行动组（Action Group）配置：
   • 点击“Add action group”。
   • Action group name:ExecutePythonCode
   • Description: “在一个安全沙箱中执行提供的Python代码并返回结果。”
   • Action group type: 选择 “Define with API schemas”。
   • Lambda function: 选择之前创建的code-executor-python函数及其PROD别名。
   • API schema: 这是定义工具接口的OpenAPI Schema。你需要提供一个JSON或YAML。这里提供一个简化示例：

     openapi: 3.0.0 info: title: Python Code Executor API version: 1.0.0 paths: /execute: post: summary: Execute Python code operationId: executePythonCode requestBody: required: true content: application/json: schema: $ref: '#/components/schemas/ExecuteRequest' responses: '200': description: Code execution result content: application/json: schema: $ref: '#/components/schemas/ExecuteResponse' components: schemas: ExecuteRequest: type: object properties: code: type: string description: The Python code snippet to execute. ExecuteResponse: type: object properties: stdout: type: string stderr: type: string returncode: type: integer

   • Bedrock会根据此Schema自动生成一个“Function definition”，描述executePythonCode这个动作，并映射到Lambda函数。
5. 知识库（可选）：本例暂不添加。如果你希望Agent能参考公司内部的代码规范文档，可以在此处连接一个包含文档的S3知识库。
6. 记忆（Memory）：保持默认启用。这允许Agent在单次对话中记住之前的上下文，例如用户之前要求生成排序代码，现在要求“优化它”。
7. 完成创建：点击“Create agent”。Bedrock需要几分钟来初始化Agent。

3.2 配置代码分析 Agent

重复类似步骤，创建第二个Agent。

• Agent name:CodeAnalyzerAgent
• Description: “分析代码逻辑、复杂度并提供对比报告。”
• Model:Claude 3 Haiku（分析任务相对简单，可用更轻量、更快的模型）。
• Instructions:

  你是一个代码审查和分析专家。你的任务是分析给定的代码片段，解释其算法逻辑、时间和空间复杂度，并对比不同代码之间的优劣。 你会收到两段代码及相关数据（如执行时间）。你需要生成一份结构清晰、语言专业的分析报告。 报告应包含：1) 各代码的独立分析，2) 横向对比，3) 适用场景建议。 只基于提供的事实和分析，不要虚构信息。

• Action Group:
  • Name:AnalyzeCode
  • Lambda function: 选择code-analyzer的PROD别名。
  • API Schema(YAML示例):

    openapi: 3.0.0 info: title: Code Analyzer API version: 1.0.0 paths: /analyze: post: summary: Analyze and compare two code snippets operationId: analyzeCode requestBody: required: true content: application/json: schema: $ref: '#/components/schemas/AnalyzeRequest' responses: '200': description: Analysis report content: application/json: schema: $ref: '#/components/schemas/AnalyzeResponse' components: schemas: AnalyzeRequest: type: object properties: code_a: type: string code_b: type: string perf_a: type: string perf_b: type: string AnalyzeResponse: type: object properties: analysis: type: string

3.3 配置监督员 Agent

这是协调整个团队的核心。

• Agent name:CodeShowSupervisor
• Description: “协调代码生成与代码分析Agent，完成完整的代码演示任务。”
• Model:Claude 3 Sonnet或Opus（需要较强的任务分解和协调能力）。
• Instructions:

  你是“代码秀”项目的总指挥。用户会提出涉及代码生成、执行和分析的复合需求。 你的工作流程是： 1. 理解用户的完整请求。 2. 将请求拆解为独立的子任务：a) 生成特定代码，b) 分析/对比代码。 3. 将代码生成类子任务委托给 `CodeGeneratorAgent`。 4. 将代码分析类子任务委托给 `CodeAnalyzerAgent`。 5. 收集并整合所有专业Agent的回复，形成一份完整、连贯的最终答案回复给用户。 如果用户的需求不明确，你需要与用户交互以澄清。 你本身不直接生成代码或进行深度分析，你的价值在于任务规划和结果合成。

• 多Agent协作配置：
  • 在Agent配置页面，找到“Multi-agent collaboration”部分。
  • 点击“Add agent”，分别添加之前创建好的CodeGeneratorAgent和CodeAnalyzerAgent。
  • 你需要为每个被协调的Agent提供一个简短的描述，帮助监督员理解何时调用谁。例如：
    • CodeGeneratorAgent: “用于生成和验证Python代码片段。”
    • CodeAnalyzerAgent: “用于分析和对比不同代码片段的逻辑与性能。”
• 行动组：监督员Agent通常不需要自己的行动组，它的工具就是其他专业Agent。但如果你有需要所有Agent共享的通用工具（如查询数据库），可以在这里配置。

4. 运行、测试与验证

配置完成后，我们进入最关键的验证环节：确保每个Agent能独立工作，并且监督员能正确协调它们。

4.1 独立测试专业Agent

在Bedrock控制台，每个Agent都有一个“Test”窗口。

1. 测试CodeGeneratorAgent:

   • 在Agent详情页，点击“Test”。
   • 输入：“写一个Python函数，计算斐波那契数列的第n项。”
   • 预期行为：Agent应生成一个Python函数（例如使用递归或循环），然后自动识别到它有一个“执行代码”的工具，并尝试调用它。由于我们的Lambda需要code参数，而生成的代码本身是确定的，Agent可能会尝试用n=10这样的默认值去执行，然后将代码和结果一并返回。
   • 查看日志：在测试窗口下方，打开“Trace”视图。你可以清晰看到Agent的推理链条（Thought）：理解请求 -> 决定生成代码 -> 决定调用工具 -> 准备参数 -> 调用Lambda -> 接收结果 -> 组织最终回复。这是排查问题的黄金位置。

2. 测试CodeAnalyzerAgent:

   • 输入：“请分析以下两段排序代码。[这里粘贴快速排序和冒泡排序的代码]”
   • 预期行为：Agent应识别出需要调用AnalyzeCode工具，并尝试从输入中提取code_a和code_b参数。由于我们没有提供性能数据，perf_a和perf_b可能为空。它应返回一份基于代码静态分析的报告。

4.2 端到端测试监督员Agent

这是检验整个系统是否成功的关键。

1. 发起复合请求：在CodeShowSupervisor的测试窗口输入：“请用Python演示快速排序和冒泡排序，并比较它们的性能。”
2. 观察多Agent协作流程：
   • 在“Trace”中，你应该看到CodeShowSupervisor的推理过程：“用户请求涉及代码生成和性能比较。我需要将任务分解。首先，让CodeGeneratorAgent生成快速排序代码并执行；其次，让CodeGeneratorAgent生成冒泡排序代码并执行；最后，将两段代码和性能数据交给CodeAnalyzerAgent进行分析。”
   • 接着，Trace会显示它调用CodeGeneratorAgent的子会话。进入子会话的Trace，你能看到代码生成Agent独立工作的完整过程。
   • 完成后，监督员会发起对CodeAnalyzerAgent的调用。
   • 最终，监督员汇总所有信息，生成类似这样的回复：

     “我已为您完成了快速排序和冒泡排序的演示与比较。

     1. 快速排序代码及结果：[生成的代码块] 执行结果：[Lambda返回的执行输出]

     2. 冒泡排序代码及结果：[生成的代码块] 执行结果：[Lambda返回的执行输出]

     3. 性能对比分析：[CodeAnalyzerAgent生成的分析报告]

     总结：...”

3. 验证输出：检查最终回复是否包含了所有预期的部分：两段可运行的代码、各自的执行结果、一份专业的对比分析。这证明多Agent协作管道是通的。

4.3 通过API进行集成测试

控制台测试通过后，便可以通过Bedrock的Runtime API将Agent集成到你的应用程序中。以下是使用AWS SDK for Python (Boto3) 调用Agent的示例：

import boto3 import json # 初始化Bedrock Agent Runtime客户端 bedrock_agent_runtime = boto3.client( service_name='bedrock-agent-runtime', region_name='us-east-1' # 替换为你的区域 ) # 监督员Agent的ID agent_id = 'YOUR_SUPERVISOR_AGENT_ID' agent_alias_id = 'TSTALIASID' # 或你创建的别名ID def invoke_agent_with_session(prompt, session_id): """ 调用Agent并维持会话。 """ try: response = bedrock_agent_runtime.invoke_agent( agentId=agent_id, agentAliasId=agent_alias_id, sessionId=session_id, # 用于维持多轮对话记忆 inputText=prompt, enableTrace=True # 启用追踪，便于调试 ) # 处理流式响应（Agent回复可能是流式的） completion = "" for event in response.get('completion', []): chunk = event.get('chunk', {}) if 'bytes' in chunk: text = chunk['bytes'].decode('utf-8') completion += text print(text, end='') # 实时输出 print() # 换行 # 获取追踪信息（需要从另一个API获取，此处简化） # 实际应处理 `response['trace']` 或调用 `get_agent_trace` API return completion except Exception as e: print(f"Error invoking agent: {e}") return None # 模拟一个多轮对话 session_id = "code-show-session-001" print("用户: 请演示快速排序和冒泡排序，并比较性能。") result1 = invoke_agent_with_session("请演示快速排序和冒泡排序，并比较性能。", session_id) print(f"\nAgent: {result1[:200]}...") # 打印前200字符 print("\n用户: 能把快速排序改成非递归的吗？") result2 = invoke_agent_with_session("能把快速排序改成非递归的吗？", session_id) # 由于有记忆，Agent应该能理解“快速排序”指的是上一轮对话中的代码，并在此基础上修改。

5. 常见问题、排查路径与最佳实践

构建和运行此类AI Agent应用时，会遇到各种问题。以下是基于经验的排查清单和优化建议。

5.1 常见问题与排查路径

5.2 生产环境最佳实践

1. 版本控制与别名：

   • 对Agent配置、Instructions、API Schema进行版本控制（如使用AWS CloudFormation或CDK）。
   • Lambda函数务必使用别名（如PROD,STAGING）供Agent调用，而不是直接使用$LATEST。这样可以在发布新版本函数时，先更新别名指向新版本进行测试，稳定后再切换，实现蓝绿部署。

2. 监控与可观测性：

   • 启用Bedrock Trace：在调用API时设置enableTrace=True，并将Trace日志发送到CloudWatch Logs或S3，用于分析Agent的决策过程和排查问题。
   • 监控Lambda指标：关注Lambda的调用次数、持续时间、错误率和并发数。为代码执行类函数设置合理的超时和内存限制。
   • 使用CloudWatch自定义指标：记录业务相关的指标，如“用户请求复杂度”、“工具调用成功率”、“最终用户满意度”（可通过后续评分反馈）。

3. 安全与权限：

   • 最小权限原则：Agent执行角色的权限应精确到所需的具体Lambda函数和S3路径，避免使用*。
   • 输入验证与清理：在Lambda函数中，务必对来自Agent的参数进行严格的验证和清理，特别是执行用户代码的沙箱环境，要防止注入攻击和资源滥用。
   • 使用Bedrock Guardrails：为Agent配置Guardrails，过滤不当的输入和输出，防止生成有害内容或泄露敏感信息。

4. 性能与成本优化：

   • 模型选型：根据任务复杂度选择合适的模型。轻量任务用Haiku，复杂推理用Sonnet或Opus。可以在不同Agent上使用不同模型。
   • 缓存：对于频繁且结果固定的工具调用（如查询静态数据），考虑在Lambda层或API Gateway层增加缓存。
   • 设置会话超时：对于非对话型应用，可以设置较短的会话超时，以控制记忆功能带来的上下文长度和成本。

5. Instructions工程：

   • 清晰具体：Instructions是Agent的“宪法”。指令要明确、无歧义。明确说明Agent的角色、目标、可用工具以及如何/何时使用它们。
   • 迭代优化：通过测试窗不断与Agent对话，发现其误解或无效行为，然后精炼Instructions。这是一个持续的过程。
   • 提供示例：在Instructions中提供一两个输入输出的示例（Few-shot Learning），能显著提升Agent对复杂任务的理解。

通过以上步骤，你不仅能够搭建起一个用于“代码秀”演示的多Agent系统，更能掌握构建企业级AI智能体应用的核心方法论。从明确业务流程、设计Agent职责，到配置工具集成、测试协作流程，再到最后的投产优化与监控，每一步都需要细致的工程化思考。Amazon Bedrock Agents 提供了一个强大的托管平台，但成功的关键仍在于对业务逻辑的深刻理解和对AI智能体工作模式的精准把握。

🚀 30+款热门AI模型一站整合，DeepSeek/GLM/Qwen 随心用，限时 5 折。 👉 点击领海量免费额度