企业云迁移真正难的地方，通常不在于把一台服务器搬到云上，而在于如何持续完成资产发现、依赖分析、基础设施即代码生成、迁移治理，以及迁移后的运营。AWS Professional Services 的实践表明，可以基于 Amazon Bedrock AgentCore 组织一套多智能体框架，让不同 AI Agent 分别处理这些环节，并把结果串成端到端的迁移流程。
这种架构的价值不只是“让模型写 Terraform”。更重要的是，它把迁移工作拆成了有边界、可审计、可复用的专业任务，从而把 IaC 开发周期从数周压缩到数分钟级别，同时保留人工审批和治理控制点。
为什么要拆成多个 Agent
一个通用 Agent 很难同时理解 CMDB、网络拓扑、应用依赖、合规策略、Terraform 模块和生产运维。将所有能力堆在一个提示词里，往往会带来三个问题：上下文过长、输出难以验证，以及出现错误时无法定位责任环节。
多智能体方案可以按迁移生命周期划分职责：
- Discovery Agent：读取资产清单、配置、日志和依赖关系，形成应用与基础设施画像。
- IaC Agent：根据目标架构和组织模板生成 Terraform、CloudFormation 或其他 IaC 文件。
- Governance Agent：检查命名、标签、成本、区域、权限和安全策略，并将结果提交审批。
- Operations Agent：在迁移完成后处理健康检查、告警分析、运行手册和变更建议。
Amazon Bedrock AgentCore 可以作为这些 Agent 的运行基础，帮助团队组织模型调用、工具访问、运行时执行和企业系统集成。实际落地时，Agent 不应直接拥有无限权限，而应通过明确的工具接口访问 AWS API、代码仓库、CMDB 和工单系统。
一条迁移任务应该怎样流转
可以把一次迁移抽象为一组带状态的工件，而不是一段不可追踪的对话：
inventory.json
-> discovery-report.json
-> target-architecture.json
-> terraform/
-> governance-report.json
-> approval
-> deployment
-> operations-runbook.md
每个阶段都应该满足三个条件：
- 输入和输出使用结构化格式，便于下一个 Agent 消费。
- 输出保留证据和决策理由，方便审计和人工复核。
- 具有幂等性，重复执行不会无意中创建重复资源或覆盖人工修改。
例如，Discovery Agent 不应只返回“这是一个三层 Web 应用”，而应输出应用 ID、发现时间、证据来源、依赖关系、置信度和待确认项。IaC Agent 也不应直接把代码部署到生产环境，而应把代码提交到分支，交给测试、策略扫描和审批流程。
可运行的多 Agent 编排示例
下面是一个不依赖 AWS 凭证的最小 Python 示例，用来演示迁移任务如何在多个专业 Agent 之间传递。它可以直接运行，也可以把各个函数替换成实际的 Bedrock AgentCore 调用、工具调用或工作流节点。
保存为 migration_pipeline.py 后运行 python migration_pipeline.py。示例中的 Agent 使用确定性逻辑模拟模型输出，接入真实环境时应增加身份认证、输入校验、策略扫描和人工审批。
from dataclasses import dataclass, field
from typing import Any, Dict, List
import json
@dataclass
class MigrationContext:
inventory: Dict[str, Any]
artifacts: Dict[str, Any] = field(default_factory=dict)
audit_log: List[str] = field(default_factory=list)
def record(self, agent: str, message: str) -> None:
self.audit_log.append(f"{agent}: {message}")
def discovery_agent(ctx: MigrationContext) -> None:
apps = ctx.inventory.get("applications", [])
report = {
"application_count": len(apps),
"applications": [
{
"name": app["name"],
"tier": app.get("tier", "unknown"),
"dependencies": app.get("dependencies", []),
"confidence": 0.95 if app.get("dependencies") else 0.70,
}
for app in apps
],
"open_questions": [
"确认生产数据库是否需要跨可用区部署",
"确认应用是否允许在目标区域使用托管服务",
],
}
ctx.artifacts["discovery_report"] = report
ctx.record("discovery", "generated discovery report")
def iac_agent(ctx: MigrationContext) -> None:
report = ctx.artifacts["discovery_report"]
resources = []
for app in report["applications"]:
resources.append({
"resource": "aws_ecs_service",
"name": app["name"].replace("-", "_"),
"desired_count": 2,
"depends_on": app["dependencies"],
})
ctx.artifacts["iac_plan"] = {
"format": "terraform-plan-input",
"resources": resources,
"requires_review": bool(report["open_questions"]),
}
ctx.record("iac", f"generated {len(resources)} infrastructure resources")
def governance_agent(ctx: MigrationContext) -> None:
plan = ctx.artifacts["iac_plan"]
violations = []
for resource in plan["resources"]:
if resource["desired_count"] < 2:
violations.append(f"{resource['name']}: desired_count must be at least 2")
ctx.artifacts["governance_report"] = {
"status": "pass" if not violations else "fail",
"violations": violations,
"approval_required": plan["requires_review"],
}
ctx.record("governance", ctx.artifacts["governance_report"]["status"])
def operations_agent(ctx: MigrationContext) -> None:
ctx.artifacts["operations_runbook"] = {
"health_checks": ["service availability", "error rate", "latency", "database connections"],
"rollback": "restore the previous deployment and verify data consistency",
"owner": "cloud-platform-team",
}
ctx.record("operations", "generated post-migration runbook")
def main() -> None:
inventory = {
"applications": [
{"name": "orders-api", "tier": "backend", "dependencies": ["orders-db"]},
{"name": "orders-db", "tier": "database", "dependencies": []},
]
}
ctx = MigrationContext(inventory=inventory)
discovery_agent(ctx)
iac_agent(ctx)
governance_agent(ctx)
if ctx.artifacts["governance_report"]["status"] == "pass":
operations_agent(ctx)
else:
ctx.record("orchestrator", "blocked deployment because governance failed")
print(json.dumps({"artifacts": ctx.artifacts, "audit_log": ctx.audit_log}, indent=2))
if __name__ == "__main__":
main()
在真实的 AgentCore 工作流中，可以把 discovery_agent、iac_agent 等函数映射为独立 Agent，把 inventory、代码仓库、Terraform 校验器和审批系统作为受控工具。编排器只负责传递任务状态和工件，不负责替每个 Agent 做所有领域判断。
从自动生成到可治理交付
迁移自动化的边界必须提前定义。适合交给 Agent 的工作包括：
- 根据标准模块生成重复性的 Terraform 资源。
- 汇总 CMDB、日志和扫描结果，形成资产画像。
- 对 IaC 运行格式化、静态检查、成本估算和策略验证。
- 根据监控指标生成运行手册初稿和常见故障处理建议。
需要人工参与或更严格保护的工作包括：
- 生产网络、身份权限和数据迁移策略的最终批准。
- 无法确认依赖关系时的架构选择。
- 涉及删除资源、修改数据或扩大权限的操作。
- 低置信度发现结果的确认。
建议为每个 Agent 设置最小权限 IAM 角色，并把“生成”和“执行”分成两个阶段。生成阶段可以创建分支和计划文件；执行阶段必须经过自动化测试、策略扫描和明确审批。对于高风险工具，可以要求 Agent 只生成命令，由人工或受控流水线执行。
落地时的检查清单
- 先定义迁移工件格式，再选择 Agent 数量。
- 给发现结果记录证据来源、置信度和待确认问题。
- 为 IaC Agent 提供组织认可的模块、命名规则和标签策略。
- 将 Terraform plan、策略扫描、成本估算和安全检查放入交付流水线。
- 使用版本化提示词、工具定义和模型配置，保证结果可回放。
- 记录每次 Agent 调用、工具参数、输出和人工决策。
- 为删除、权限变更和生产部署设置人工审批或双重确认。
- 用成功率、返工率、人工审批耗时和迁移后故障率衡量实际收益。
多智能体架构并不会自动消除云迁移的复杂性。它的作用是把复杂工作拆成可验证的专业步骤，并让重复劳动变得更快。当 AgentCore 与结构化工件、最小权限、自动化测试和人工治理结合起来时，企业才能在扩大迁移规模的同时控制风险，而不是把风险一起自动化。