1. 项目概述当基础设施交付变成“点单式”操作你有没有经历过这样的场景业务线凌晨三点发来紧急需求——“新环境明天上线要三套测试集群、一套预发、一套灰度配置必须和生产完全一致”运维同学立刻拉起会议翻出尘封半年的 Terraform 模块文档一边查变量名一边改main.tf中间发现某个模块版本不兼容又得切分支、打补丁、跑 plan等所有配置对齐、审批走完、apply 完成天已经亮了而业务方反馈“需求临时调整灰度环境规格要翻倍”。这种循环在过去三年里我参与过的七个中型以上系统交付项目中平均每个环境交付耗时 3.7 天其中 68% 的时间花在人工校验、跨团队对齐、配置拼写纠错和反复重试上。而标题里说的“四个智能体把 IaC 交付从数周压到分钟级”不是营销话术是我们在一个真实模拟项目X中落地验证后的实测结果——从输入自然语言需求比如“给我一个带自动扩缩容的 Python Web 应用环境前端用 Cloudflare后端跑在 Kubernetes 上数据库用高可用 PostgreSQL所有资源加标签 ownerdev-team-a”到全部云资源创建完成、健康检查通过、CI/CD 流水线自动注入全程耗时 4 分 23 秒且零人工干预。核心不在“快”而在“稳”每次交付的 Terraform 状态文件、执行日志、资源拓扑图、安全合规检查报告全部自动生成并归档可回溯、可审计、可复现。它解决的从来不是“能不能自动化”的问题而是“如何让非基础设施背景的人也能安全、精准、可追溯地触发基础设施变更”的问题。适合正在被 IaC 落地卡住脖子的 SRE 团队、想摆脱“配置即代码”黑盒困境的 DevOps 工程师以及需要快速验证多套架构方案的产品技术负责人——你不需要成为 Terraform 专家但你需要知道当你说“我要一个能扛住 5000 QPS 的 API 网关”系统就真的只给你建那个网关不多也不少且每一步都经得起推敲。2. 整体设计思路与四大智能体协同逻辑2.1 为什么是“四个”而不是一个大模型或一个 Agent这是整个方案最常被问错的问题。很多人第一反应是“直接丢给一个超大模型让它读完所有云文档、写完所有 TF 文件、再执行 apply 不就行了” 实际踩过坑之后你会发现这种“单体智能体”在 IaC 场景下是灾难性的。我试过用 128K 上下文的模型直接解析 AWS 官方 Terraform Provider 文档近 2000 页 Markdown让它生成一个带 EKS 节点组自动扩缩容的完整模块。结果是它正确引用了aws_eks_node_group资源但把capacity_type ON_DEMAND错写成ONDEMAND少了个下划线导致terraform plan直接报错更致命的是它把desired_capacity和max_size的数值逻辑搞反了生成的配置会让节点组在低负载时疯狂扩容。这不是模型能力问题而是任务本质决定的——IaC 交付是一个强状态、多阶段、高容错门槛的工程流水线它天然需要职责分离。我们最终采用的四智能体架构是严格按“人类工程师日常协作模式”逆向拆解出来的需求理解智能体Requirement Interpreter像一位资深售前顾问专精于把模糊、口语化、带业务语义的输入如“要个能抗住大促的订单库”翻译成精确、无歧义、带约束条件的技术需求清单。它不碰代码只输出结构化 JSON字段包括service_type,scale_expectation,availability_requirement,security_class,compliance_tags。架构设计智能体Architecture Designer像一位首席架构师接收上一环节的 JSON结合当前云平台能力图谱我们维护了一份动态更新的“能力知识库”含各 Region 支持的服务、最新 Terraform Provider 版本、已验证的模块组合模式输出符合最佳实践的组件拓扑图Mermaid 格式文本和模块选型建议例如“推荐使用 community 模块terraform-aws-eksv23.0而非官方aws_eks_cluster原生资源因前者内置了 IRSA 配置和节点组滚动更新策略”。代码生成智能体Code Generator像一位资深 Terraform 工程师它只做一件事——根据前两步输出的“需求清单 架构图 模块建议”调用本地缓存的、经过千次 CI 验证的模块模板库填充变量、注入合规标签、插入监控告警钩子生成可直接terraform init plan的代码包。它不推理、不创造只做高精度模板装配。执行与验证智能体Executor Verifier像一位严谨的 QA 工程师兼值班 SRE它不信任任何上游输出。它会先运行terraform validate和tflint进行静态检查再启动一个隔离沙箱环境执行terraform plan -outplan.binary并解析输出比对资源类型、数量、关键参数是否与需求清单一致最后才执行terraform apply并在资源创建后主动调用云平台 API如describe-cluster,get-db-instance做终态验证只有全部健康检查通过才标记交付成功。这四个角色之间通过轻量级消息总线我们用的是本地 Redis Stream传递结构化载荷每个环节失败都会触发熔断返回具体错误码和修复建议例如“架构设计失败region us-west-1 当前不支持 Graviton3 实例类型建议切换至 us-east-1 或降级为 Graviton2”而不是让错误一路滚到底层执行层再崩溃。这种设计把“大模型的创造性”和“工程系统的确定性”做了物理隔离——模型负责理解与设计代码与执行负责保底与验证。2.2 Bedrock AgentCore 为何成为底层基座它解决了什么真问题提到 Bedrock AgentCore很多人的第一反应是“又是 AWS 新玩具是不是又要学一堆新概念” 其实不然。AgentCore 的核心价值恰恰在于它刻意不做的事情它不提供开箱即用的“智能体模板”不封装特定云服务 SDK甚至不强制你用 Lambda 做后端。它本质上是一个“智能体运行时契约框架”定义了一套标准化的输入/输出协议、状态管理接口和工具调用规范。我们选择它的根本原因有三个全是血泪教训换来的第一工具调用的确定性保障。在早期用 LangChain 自研 Agent 时我们遇到最头疼的问题是当模型决定调用get_aws_region_list()工具后如果该工具因网络抖动返回空数组模型会“脑补”出一个不存在的 region比如us-west-9然后继续往下走直到terraform init报错。AgentCore 强制要求每个工具必须声明required_params、optional_params和error_handling_strategy如retry_on_timeout,fallback_to_default并且所有工具调用结果必须经过 Schema 校验才能进入下一步。这意味着当get_aws_region_list()返回空时AgentCore 会直接中断流程抛出TOOL_EXECUTION_FAILED错误而不是让模型瞎猜。第二状态持久化的工程友好性。IaC 交付不是一次对话而是一个可能持续数小时的长周期任务尤其涉及跨账号资源、审批流集成时。AgentCore 内置的 State Manager 支持将任意 JSON 状态快照存入 DynamoDB并提供get_state(),update_state(),delete_state()接口。我们利用这个能力把每个交付任务的中间产物——原始需求文本、解析后的 JSON 清单、生成的 Terraform 代码哈希、plan 输出摘要、最终资源 ID 列表——全部存为状态快照。这带来的好处是当某次交付因网络中断失败运维同学只需输入任务 ID系统就能从断点恢复而不是从头再来。更重要的是这些状态快照天然成为审计线索你可以随时查询“ID-78921 在 2024-05-12 14:23:01 生成的代码其aws_s3_bucket资源的acl参数值是多少”。第三与现有 CI/CD 的无缝缝合能力。AgentCore 不要求你把所有东西都搬到它的世界里。我们的 Terraform 代码仓库依然托管在 GitLabCI 流水线依然是 Jenkins。我们只是在 Jenkins Pipeline 的某个 stage 中用curl调用 AgentCore 的/invokeAPI传入需求 JSON 和指定的智能体 ID如arch-designer-v2然后等待它返回结构化结果。AgentCore 的输出是纯 JSONJenkins 可以直接解析并写入环境变量后续 stage 就能用${ARCH_OUTPUT_MODULE_NAME}来动态 checkout 对应的 Terraform 模块。这种“API 优先”的集成方式让我们避免了推翻重来旧系统照常运转新能力平滑注入。2.3 MCPModel Control Protocol不是新协议而是工程控制论的回归MCP 这个词最近很热但很多文章把它讲成了玄学。在我实际落地过程中MCP 的本质非常朴素它是一套防止大模型“越权发挥”的刹车系统。它的核心思想来自经典控制论中的“负反馈调节”——不是靠模型自己判断“我该不该做这件事”而是由外部系统实时监控模型行为一旦偏离预设轨道立即干预。我们实现的 MCP 包含三个硬性控制层输入层过滤器Input Sanitizer所有用户输入在进入模型前必须经过正则规则 语义白名单双重校验。例如禁止出现rm -rf,destroy,force-unlock等高危关键词同时对业务术语做映射如用户说“删掉旧库”会被重写为“将数据库实例prod-db-old的deletion_protection设置为 false并安排在维护窗口期执行销毁”确保模型永远只看到“建设性指令”。输出层拦截器Output Validator模型生成的任何代码、命令、配置片段在落盘或执行前必须通过三层校验① Terraform HCL 语法校验terraform fmt -check② 安全策略扫描我们集成了开源版 Checkov规则集包含 127 条自定义规则如“S3 存储桶必须启用服务器端加密”、“EC2 实例禁止使用密钥对登录必须用 Session Manager”③ 合规性签名所有生成的.tf文件末尾自动追加注释# GENERATED_BY_MCP_v1.2 | SIGNATURE: sha256:xxx供审计系统追踪。执行层熔断器Execution Circuit Breaker这是最狠的一环。我们给每个智能体的执行设置了“信用额度”。例如代码生成智能体单次调用最多允许生成 500 行代码、调用 3 个外部工具、耗时不超过 90 秒。一旦超限MCP 熔断器会立即终止进程返回RATE_LIMIT_EXCEEDED错误并触发告警。这个机制直接杜绝了模型“发散式创作”——它不会因为想“优化体验”而擅自给你加一堆没用的 CloudWatch 告警也不会因为“觉得配置不够酷”而引入一个未经验证的第三方 provider。MCP 不是给模型加锁而是给工程流程装上仪表盘。它让“智能”变得可测量、可干预、可归责。当你看到监控面板上显示“今日 MCP 拦截高危操作 17 次其中 12 次为未授权的资源销毁请求”你就知道这套系统真正开始守护你的生产环境了。3. 核心细节解析与实操要点3.1 需求理解智能体如何把“老板一句话”变成机器可执行的 JSON很多团队卡在第一步用户输入太随意模型一解析就失真。比如业务方说“搞个能跑 AI 模型的环境要快别太贵。” 这句话里藏着至少五个模糊点“AI 模型”指训练还是推理“快”是启动快、计算快还是网络延迟低“别太贵”是预算 5000 还是 50000我们解决这个问题靠的不是更强的模型而是结构化提示工程 领域词典 主动澄清机制。首先我们构建了一个轻量级的“云服务语义词典”它不是大而全的百科而是聚焦高频业务场景的映射表。例如当用户提到“AI 模型”词典会将其归类为inference_workload或training_workload并关联默认硬件偏好inference_workload → g5.xlarge,training_workload → p4d.24xlarge“快”被拆解为三个维度startup_time 60s对应 Spot 实例或 Serverless、compute_latency 10ms对应 GPU 实例、network_latency 50ms对应同 Region 部署“别太贵”被映射为预算等级budget_tier: medium对应月均 $2000-$5000并关联成本优化策略如“强制启用 Savings Plans”、“禁用按需实例”。其次我们设计了分阶段提示Multi-step Prompting。模型不是一次性输出最终 JSON而是分三步走意图识别先判断用户输入属于哪类需求Web App / Data Pipeline / AI Inference / Legacy Migration若置信度 0.85则触发澄清问题如“请问这个 AI 环境主要用于模型训练还是在线推理”约束提取在确认意图后专门抽取显性约束如“要 3 个副本”、“必须用 Aurora”和隐性约束如“金融行业”→ 自动添加compliance_tags: [PCI-DSS, SOC2]JSON 生成最后才生成严格遵循 Schema 的输出Schema 定义如下精简版{ service_type: string, scale_expectation: { qps: number, concurrent_users: number, data_volume_gb_per_day: number }, availability_requirement: string, // single_az, multi_az, global_active security_class: string, // public_facing, internal_only, pci_dss compliance_tags: [string], budget_tier: string, // low, medium, high preferred_region: string }提示这个 Schema 不是固定死的。我们每周会从历史交付日志中提取“用户追问高频问题”动态扩充 Schema 字段。例如上周发现 23% 的需求中用户会追问“备份策略”我们就新增了backup_policy对象包含retention_days,point_in_time_recovery_enabled等子字段。这种“数据驱动的 Schema 演进”让智能体越来越懂业务语言。3.2 架构设计智能体如何让“画图”不再拍脑袋架构设计环节最容易沦为“PPT 架构师”——画一堆漂亮的框但落地时发现 A 服务不支持 B 协议C 组件在 D Region 没有可用区。我们的解法是把架构决策变成一场“能力匹配游戏”。我们维护了一个名为cloud-capability-kb.json的知识库它不是静态文档而是一个可查询的、带版本号的 JSON 数据库。每条记录代表一个“云能力单元”结构如下{ id: aws-eks-autoscaling-v2, provider: aws, service: eks, feature: node_group_autoscaling, supported_regions: [us-east-1, us-west-2, ap-northeast-1], min_provider_version: 4.65.0, recommended_module: terraform-aws-eks, module_version_range: 22.0.0, 24.0.0, known_limitations: [不支持混合实例策略与 Fargate 同时启用], last_verified_at: 2024-05-10T08:22:15Z }当架构设计智能体收到需求如{service_type: ai_inference, scale_expectation: {qps: 5000}}它会执行以下步骤能力检索查询知识库找出所有匹配service_typeai_inference且qps 5000的能力单元得到候选列表如aws-eks-autoscaling-v2,aws-ecs-fargate-spot-v3,aws-lambda-provisioned-concurrency-v1约束过滤根据需求中的availability_requirement和security_class剔除不满足的选项如security_classpci_dss会过滤掉所有未通过 PCI 认证的 Fargate 配置成本排序调用云厂商的 Pricing Calculator API我们封装了 AWS/Azure/GCP 的统一接口计算每个候选方案的预估月成本按budget_tier过滤medium预算只保留 $1500-$4000 区间拓扑生成选定最优方案后不是手动画图而是调用预置的 Mermaid 模板引擎。例如选中aws-eks-autoscaling-v2就加载eks-inference-topology.mmd模板将占位符{{cluster_name}},{{node_count}}替换为实际值输出标准 Mermaid 代码。注意我们严禁智能体“自由发挥”画图。所有拓扑图必须源自经过验证的模板库模板由 SRE 团队每月评审更新。曾有一次模型试图“优化”一个 Kafka 集群图把 ZooKeeper 组件删掉了认为它过时结果生成的图无法部署。从此我们立下铁律架构图 模板 参数绝不允许模型修改拓扑结构本身。3.3 代码生成智能体为什么“抄作业”比“写作文”更可靠这是最受质疑的一环“让模型写代码不是更酷吗” 我的答案是在 IaC 领域“酷”等于事故。我们坚持“代码生成 高精度模板装配”原因有三可审计性每一行生成的代码都能追溯到模板库中的具体 commit ID 和行号。当安全团队要求审查“S3 存储桶的加密配置”我们直接给出templates/aws/s3-bucket-encrypted.tfv3.2.1#L45-L52而不是“模型根据上下文生成的”。可测试性模板库中的每个.tf文件都配有对应的test.tfvars和expected-output.json每天凌晨自动跑terraform test。新模板入库前必须通过 100% 的测试用例。可演进性当 AWS 发布新功能如 EKS 的 Bottlerocket 支持我们不是让模型去学而是 SRE 工程师写好新模板更新知识库智能体下次调用时自动切换。我们的模板库采用“分层继承”设计基础层base定义通用变量、Provider 配置、Backend 设置服务层service如eks-cluster,rds-postgresql,alb-application每个模板专注一个服务通过count和for_each支持多实例场景层scenario如ai-inference-stack,payment-gateway-stack它不写新资源而是组合服务层模板并注入场景特有逻辑如支付网关模板会自动添加 PCI 合规检查模块。代码生成智能体的工作流极其简单接收架构设计输出含recommended_module和module_version_range查询模板库找到匹配的场景模板如ai-inference-stack解析需求 JSON将scale_expectation.qps映射为eks_node_group_desired_capacity将compliance_tags注入所有资源的tags块执行terraform fmt格式化生成最终.tf文件。实操心得我们给每个模板配了一个README.md里面明确写着“适用场景”、“不适用场景”、“已知问题”。例如rds-postgresql模板的 README 第一行就写“⚠️ 本模板不适用于需要跨区域读副本的场景请改用rds-global-database模板”。这种“诚实的文档”比任何模型幻觉都管用。3.4 执行与验证智能体为什么“执行”才是最危险的环节很多人以为生成代码就结束了其实真正的风险在执行层。我们统计过过去一年中 73% 的 IaC 生产事故发生在terraform apply阶段而非代码生成阶段。原因很简单代码是静态的环境是动态的。terraform apply可能遭遇云 API 限流、资源配额不足、依赖服务临时不可用、甚至 Terraform Provider 本身的 bug。我们的执行与验证智能体不是简单地system(terraform apply)而是一个带七重防护的“手术台”沙箱预检Sandbox Pre-check在真实环境执行前先启动一个隔离的 Docker 容器挂载生成的代码和terraform.tfstate运行terraform init -backendfalse和terraform validate。这一步能捕获 90% 的语法错误和变量缺失。Plan 安全校验Plan Security Scanterraform plan -outplan.binary生成二进制计划后我们不直接执行而是用terraform show -json plan.binary解析为 JSON然后逐条检查是否有create操作涉及aws_iam_role且assume_role_policy包含Principal: {Service: ec2.amazonaws.com}这可能是提权风险是否有aws_s3_bucket的acl字段值为public-read所有aws_security_group_rule的cidr_blocks是否包含0.0.0.0/0。这些检查由 Checkov 的自定义规则驱动失败则熔断。资源拓扑比对Topology Diff将 Plan JSON 中的资源列表与需求 JSON 中的scale_expectation做数学比对。例如需求要求qps: 5000而 Plan 显示只创建了 2 个aws_eks_node_group每个desired_capacity: 2则总容量为 4低于需求自动拒绝执行。终态健康检查Post-deploy Health Checkapply成功后不立即返回“成功”而是启动一个轮询任务对 EKS 集群调用aws eks describe-cluster --name xxx检查status ACTIVE对 RDS 实例调用aws rds describe-db-instances --db-instance-identifier xxx检查DBInstanceStatus available且StorageEncrypted true对 ALB调用aws elbv2 describe-load-balancers --names xxx检查State.Code active。所有检查通过才标记交付完成。关键技巧我们给每个健康检查设置了“容忍窗口”。例如EKS 集群从CREATING到ACTIVE平均耗时 12 分钟但我们设置的超时是 25 分钟并且每 90 秒检查一次。这样既给了云平台足够时间又不会无限等待。这个窗口值不是拍脑袋而是从过去 30 天的交付日志中用 P95 分位数计算出来的。4. 实操过程与核心环节实现4.1 环境准备从零搭建四智能体流水线整个系统基于 AWS 云原生服务构建但所有组件均可替换为其他云厂商对应服务如 Azure 的 Logic Apps 替代 Step FunctionsGCP 的 Cloud Run 替代 ECS。以下是核心组件清单与部署逻辑组件选型理由关键配置AgentCore Runtime使用 ECS Fargate免运维、弹性伸缩CPU: 2 vCPU, Memory: 4GB, Log Group:/ecs/agentcoreRedis Stream作为智能体间消息总线低延迟、高吞吐Engine: Redis 7.0, Replicas: 1, Encryption at rest: enabledDynamoDB State Table存储任务状态支持高并发读写Billing mode: On-demand, TTL attribute:expires_atTerraform Backend使用 S3 DynamoDB Lock保证状态一致性S3 Bucket:iac-state-store-prod, DynamoDB Table:terraform-state-lockPricing Calculator API封装 AWS/Azure/GCP 定价接口统一调用部署在 Lambda冷启动优化预留并发 5内存 1024MB部署流程分四步全部通过 Terraform 代码化基础设施层用根模块创建 VPC、Subnet、Security Group、IAM Roles含AgentCoreExecutionRole权限最小化仅允许调用lambda:InvokeFunction,dynamodb:GetItem,s3:GetObject数据层创建 Redis Cluster 和 DynamoDB Table其中 DynamoDB 的Partition Key设为task_idSort Key设为step_id如req-parse,arch-design,code-gen方便按任务查询全流程状态运行时层部署 ECS Cluster 和 Task Definition其中containerDefinitions指定 AgentCore 的 Docker 镜像我们基于官方镜像构建预装了aws-cli,terraform,tflint,checkov智能体层为每个智能体创建独立的 ECS Service配置不同的ENTRYPOINT如需求理解智能体的 ENTRYPOINT 是python /app/interpreter.py并通过 ECS Service Discovery 实现服务发现。提示我们为每个智能体 Service 配置了独立的 Auto Scaling Policy但策略依据不是 CPU而是 Redis Stream 的 Pending Entries 数量。当req-interpreter-stream的 pending count 10就扩容 interpreter Service 的 Task Count。这种“事件驱动的弹性”比传统指标更精准。4.2 智能体开发以“需求理解智能体”为例的完整代码结构我们用 Python 开发所有智能体框架基于 FastAPI轻量、异步、OpenAPI 原生支持。以需求理解智能体interpreter.py为例其核心结构如下# interpreter.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel import re import json from typing import Dict, Any app FastAPI(titleRequirement Interpreter) class RequirementInput(BaseModel): raw_text: str session_id: str # 用于关联用户会话 class RequirementOutput(BaseModel): service_type: str scale_expectation: Dict[str, float] availability_requirement: str security_class: str compliance_tags: list budget_tier: str preferred_region: str # 加载语义词典从 S3 下载缓存到内存 SEMANTIC_DICT load_from_s3(s3://iac-kb/semantic-dict.json) app.post(/interpret, response_modelRequirementOutput) async def interpret_requirement(input: RequirementInput): try: # 步骤1输入清洗 cleaned_text clean_input(input.raw_text) # 步骤2意图识别调用微调的小模型非大模型 intent predict_intent(cleaned_text) # 模型输出web_app, ai_inference, data_pipeline... if not intent or confidence_score 0.85: raise HTTPException(status_code400, detailIntent unclear, please clarify) # 步骤3约束提取正则 词典匹配 constraints extract_constraints(cleaned_text, SEMANTIC_DICT) # 步骤4JSON 生成调用 LLM但 prompt 严格限定为 JSON 输出 result_json generate_structured_output(cleaned_text, constraints, intent) # 步骤5Schema 校验 validated RequirementOutput.parse_obj(result_json) # 步骤6存入 DynamoDB 状态表 save_to_dynamodb(input.session_id, req-parse, validated.dict()) return validated except Exception as e: save_to_dynamodb(input.session_id, req-parse, {error: str(e)}) raise HTTPException(status_code500, detailfInterpretation failed: {e}) def clean_input(text: str) - str: # 移除特殊字符标准化空格转换为小写 return re.sub(r[^\w\s\-], , text).strip().lower() def predict_intent(text: str) - str: # 调用微调的 DistilBERT 模型12MB部署在 SageMaker Real-time Endpoint # 输入text输出intent_label, confidence_score pass def extract_constraints(text: str, dict_data: dict) - dict: # 遍历语义词典匹配关键词填充约束 constraints {} for keyword, mapping in dict_data.items(): if keyword in text: constraints.update(mapping) return constraints def generate_structured_output(text: str, constraints: dict, intent: str) - dict: # 调用 Bedrock 的 Claude 3 Haiku低成本、高响应速度 # Prompt 模板 # # 你是一个专业的云架构需求分析师。请将以下用户需求严格按 JSON Schema 输出不要任何额外文字。 # 用户需求{text} # 已识别意图{intent} # 已提取约束{constraints} # JSON Schema: {RequirementOutput.schema_json()} # pass关键细节我们没有用一个大模型干所有事而是“小模型做意图识别 大模型做 JSON 生成”。因为意图识别是分类任务DistilBERT 就够了准确率 92%延迟 200ms而 JSON 生成需要强推理交给 Claude 3 Haiku它在 1000 token 内的 JSON 生成准确率高达 98.7%。这种“大小模型协同”比单一大模型更稳、更快、更便宜。4.3 MCP 控制层实现三道防线的代码级落地MCP 的三道防线全部通过中间件Middleware方式注入到 FastAPI 应用中与业务逻辑解耦# mcp_middleware.py from fastapi import Request, Response, HTTPException from starlette.middleware.base import BaseHTTPMiddleware import hashlib import time class InputSanitizerMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): # 读取请求体 body await request.body() text body.decode(utf-8) # 正则过滤高危词 dangerous_patterns [rrm\s-rf, rdestroy, rforce-unlock, r--auto-approve] for pattern in dangerous_patterns: if re.search(pattern, text, re.IGNORECASE): raise HTTPException(status_code400, detailInput contains prohibited command) # 业务术语映射 term_mapping { 删掉旧库: set deletion_protectionfalse for prod-db-old, 关掉监控: disable CloudWatch alarms for resource X } for src, dst in term_mapping.items(): text text.replace(src, dst) # 重写请求体 request._body text.encode(utf-8) return await call_next(request) class OutputValidatorMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): response await call_next(request) # 拦截响应体假设是 JSON if response.headers.get(content-type) application/json: body b async for chunk in response.body_iterator: body chunk try: data json.loads(body.decode(utf-8)) # 如果是代码生成响应执行安全扫描 if generated_code in data: scan_result run_checkov_scan(data[generated_code]) if not scan_result[passed]: raise HTTPException(status_code422, detailfSecurity scan failed: {scan_result[issues]}) # 添加 MCP 签名 signature hashlib.sha256(body).hexdigest()[:16] data[mcp_signature] fsha256:{signature} response.body_iterator iter([json.dumps(data, ensure_asciiFalse).encode(utf-8)]) except Exception as e: raise HTTPException(status_code500, detailfOutput validation error: {e}) return response class ExecutionCircuitBreaker: def __init__(self): self.credit_pool {} def check_credit(self, 
长河君 · 2026/10/22 16:15:32 · 阅读 1.2k 四智能体驱动的IaC交付：从自然语言到云环境分钟级生成
1. 项目概述当基础设施交付变成“点单式”操作你有没有经历过这样的场景业务线凌晨三点发来紧急需求——“新环境明天上线要三套测试集群、一套预发、一套灰度配置必须和生产完全一致”运维同学立刻拉起会议翻出尘封半年的 Terraform 模块文档一边查变量名一边改main.tf中间发现某个模块版本不兼容又得切分支、打补丁、跑 plan等所有配置对齐、审批走完、apply 完成天已经亮了而业务方反馈“需求临时调整灰度环境规格要翻倍”。这种循环在过去三年里我参与过的七个中型以上系统交付项目中平均每个环境交付耗时 3.7 天其中 68% 的时间花在人工校验、跨团队对齐、配置拼写纠错和反复重试上。而标题里说的“四个智能体把 IaC 交付从数周压到分钟级”不是营销话术是我们在一个真实模拟项目X中落地验证后的实测结果——从输入自然语言需求比如“给我一个带自动扩缩容的 Python Web 应用环境前端用 Cloudflare后端跑在 Kubernetes 上数据库用高可用 PostgreSQL所有资源加标签 ownerdev-team-a”到全部云资源创建完成、健康检查通过、CI/CD 流水线自动注入全程耗时 4 分 23 秒且零人工干预。核心不在“快”而在“稳”每次交付的 Terraform 状态文件、执行日志、资源拓扑图、安全合规检查报告全部自动生成并归档可回溯、可审计、可复现。它解决的从来不是“能不能自动化”的问题而是“如何让非基础设施背景的人也能安全、精准、可追溯地触发基础设施变更”的问题。适合正在被 IaC 落地卡住脖子的 SRE 团队、想摆脱“配置即代码”黑盒困境的 DevOps 工程师以及需要快速验证多套架构方案的产品技术负责人——你不需要成为 Terraform 专家但你需要知道当你说“我要一个能扛住 5000 QPS 的 API 网关”系统就真的只给你建那个网关不多也不少且每一步都经得起推敲。2. 整体设计思路与四大智能体协同逻辑2.1 为什么是“四个”而不是一个大模型或一个 Agent这是整个方案最常被问错的问题。很多人第一反应是“直接丢给一个超大模型让它读完所有云文档、写完所有 TF 文件、再执行 apply 不就行了” 实际踩过坑之后你会发现这种“单体智能体”在 IaC 场景下是灾难性的。我试过用 128K 上下文的模型直接解析 AWS 官方 Terraform Provider 文档近 2000 页 Markdown让它生成一个带 EKS 节点组自动扩缩容的完整模块。结果是它正确引用了aws_eks_node_group资源但把capacity_type ON_DEMAND错写成ONDEMAND少了个下划线导致terraform plan直接报错更致命的是它把desired_capacity和max_size的数值逻辑搞反了生成的配置会让节点组在低负载时疯狂扩容。这不是模型能力问题而是任务本质决定的——IaC 交付是一个强状态、多阶段、高容错门槛的工程流水线它天然需要职责分离。我们最终采用的四智能体架构是严格按“人类工程师日常协作模式”逆向拆解出来的需求理解智能体Requirement Interpreter像一位资深售前顾问专精于把模糊、口语化、带业务语义的输入如“要个能抗住大促的订单库”翻译成精确、无歧义、带约束条件的技术需求清单。它不碰代码只输出结构化 JSON字段包括service_type,scale_expectation,availability_requirement,security_class,compliance_tags。架构设计智能体Architecture Designer像一位首席架构师接收上一环节的 JSON结合当前云平台能力图谱我们维护了一份动态更新的“能力知识库”含各 Region 支持的服务、最新 Terraform Provider 版本、已验证的模块组合模式输出符合最佳实践的组件拓扑图Mermaid 格式文本和模块选型建议例如“推荐使用 community 模块terraform-aws-eksv23.0而非官方aws_eks_cluster原生资源因前者内置了 IRSA 配置和节点组滚动更新策略”。代码生成智能体Code Generator像一位资深 Terraform 工程师它只做一件事——根据前两步输出的“需求清单 架构图 模块建议”调用本地缓存的、经过千次 CI 验证的模块模板库填充变量、注入合规标签、插入监控告警钩子生成可直接terraform init plan的代码包。它不推理、不创造只做高精度模板装配。执行与验证智能体Executor Verifier像一位严谨的 QA 工程师兼值班 SRE它不信任任何上游输出。它会先运行terraform validate和tflint进行静态检查再启动一个隔离沙箱环境执行terraform plan -outplan.binary并解析输出比对资源类型、数量、关键参数是否与需求清单一致最后才执行terraform apply并在资源创建后主动调用云平台 API如describe-cluster,get-db-instance做终态验证只有全部健康检查通过才标记交付成功。这四个角色之间通过轻量级消息总线我们用的是本地 Redis Stream传递结构化载荷每个环节失败都会触发熔断返回具体错误码和修复建议例如“架构设计失败region us-west-1 当前不支持 Graviton3 实例类型建议切换至 us-east-1 或降级为 Graviton2”而不是让错误一路滚到底层执行层再崩溃。这种设计把“大模型的创造性”和“工程系统的确定性”做了物理隔离——模型负责理解与设计代码与执行负责保底与验证。2.2 Bedrock AgentCore 为何成为底层基座它解决了什么真问题提到 Bedrock AgentCore很多人的第一反应是“又是 AWS 新玩具是不是又要学一堆新概念” 其实不然。AgentCore 的核心价值恰恰在于它刻意不做的事情它不提供开箱即用的“智能体模板”不封装特定云服务 SDK甚至不强制你用 Lambda 做后端。它本质上是一个“智能体运行时契约框架”定义了一套标准化的输入/输出协议、状态管理接口和工具调用规范。我们选择它的根本原因有三个全是血泪教训换来的第一工具调用的确定性保障。在早期用 LangChain 自研 Agent 时我们遇到最头疼的问题是当模型决定调用get_aws_region_list()工具后如果该工具因网络抖动返回空数组模型会“脑补”出一个不存在的 region比如us-west-9然后继续往下走直到terraform init报错。AgentCore 强制要求每个工具必须声明required_params、optional_params和error_handling_strategy如retry_on_timeout,fallback_to_default并且所有工具调用结果必须经过 Schema 校验才能进入下一步。这意味着当get_aws_region_list()返回空时AgentCore 会直接中断流程抛出TOOL_EXECUTION_FAILED错误而不是让模型瞎猜。第二状态持久化的工程友好性。IaC 交付不是一次对话而是一个可能持续数小时的长周期任务尤其涉及跨账号资源、审批流集成时。AgentCore 内置的 State Manager 支持将任意 JSON 状态快照存入 DynamoDB并提供get_state(),update_state(),delete_state()接口。我们利用这个能力把每个交付任务的中间产物——原始需求文本、解析后的 JSON 清单、生成的 Terraform 代码哈希、plan 输出摘要、最终资源 ID 列表——全部存为状态快照。这带来的好处是当某次交付因网络中断失败运维同学只需输入任务 ID系统就能从断点恢复而不是从头再来。更重要的是这些状态快照天然成为审计线索你可以随时查询“ID-78921 在 2024-05-12 14:23:01 生成的代码其aws_s3_bucket资源的acl参数值是多少”。第三与现有 CI/CD 的无缝缝合能力。AgentCore 不要求你把所有东西都搬到它的世界里。我们的 Terraform 代码仓库依然托管在 GitLabCI 流水线依然是 Jenkins。我们只是在 Jenkins Pipeline 的某个 stage 中用curl调用 AgentCore 的/invokeAPI传入需求 JSON 和指定的智能体 ID如arch-designer-v2然后等待它返回结构化结果。AgentCore 的输出是纯 JSONJenkins 可以直接解析并写入环境变量后续 stage 就能用${ARCH_OUTPUT_MODULE_NAME}来动态 checkout 对应的 Terraform 模块。这种“API 优先”的集成方式让我们避免了推翻重来旧系统照常运转新能力平滑注入。2.3 MCPModel Control Protocol不是新协议而是工程控制论的回归MCP 这个词最近很热但很多文章把它讲成了玄学。在我实际落地过程中MCP 的本质非常朴素它是一套防止大模型“越权发挥”的刹车系统。它的核心思想来自经典控制论中的“负反馈调节”——不是靠模型自己判断“我该不该做这件事”而是由外部系统实时监控模型行为一旦偏离预设轨道立即干预。我们实现的 MCP 包含三个硬性控制层输入层过滤器Input Sanitizer所有用户输入在进入模型前必须经过正则规则 语义白名单双重校验。例如禁止出现rm -rf,destroy,force-unlock等高危关键词同时对业务术语做映射如用户说“删掉旧库”会被重写为“将数据库实例prod-db-old的deletion_protection设置为 false并安排在维护窗口期执行销毁”确保模型永远只看到“建设性指令”。输出层拦截器Output Validator模型生成的任何代码、命令、配置片段在落盘或执行前必须通过三层校验① Terraform HCL 语法校验terraform fmt -check② 安全策略扫描我们集成了开源版 Checkov规则集包含 127 条自定义规则如“S3 存储桶必须启用服务器端加密”、“EC2 实例禁止使用密钥对登录必须用 Session Manager”③ 合规性签名所有生成的.tf文件末尾自动追加注释# GENERATED_BY_MCP_v1.2 | SIGNATURE: sha256:xxx供审计系统追踪。执行层熔断器Execution Circuit Breaker这是最狠的一环。我们给每个智能体的执行设置了“信用额度”。例如代码生成智能体单次调用最多允许生成 500 行代码、调用 3 个外部工具、耗时不超过 90 秒。一旦超限MCP 熔断器会立即终止进程返回RATE_LIMIT_EXCEEDED错误并触发告警。这个机制直接杜绝了模型“发散式创作”——它不会因为想“优化体验”而擅自给你加一堆没用的 CloudWatch 告警也不会因为“觉得配置不够酷”而引入一个未经验证的第三方 provider。MCP 不是给模型加锁而是给工程流程装上仪表盘。它让“智能”变得可测量、可干预、可归责。当你看到监控面板上显示“今日 MCP 拦截高危操作 17 次其中 12 次为未授权的资源销毁请求”你就知道这套系统真正开始守护你的生产环境了。3. 核心细节解析与实操要点3.1 需求理解智能体如何把“老板一句话”变成机器可执行的 JSON很多团队卡在第一步用户输入太随意模型一解析就失真。比如业务方说“搞个能跑 AI 模型的环境要快别太贵。” 这句话里藏着至少五个模糊点“AI 模型”指训练还是推理“快”是启动快、计算快还是网络延迟低“别太贵”是预算 5000 还是 50000我们解决这个问题靠的不是更强的模型而是结构化提示工程 领域词典 主动澄清机制。首先我们构建了一个轻量级的“云服务语义词典”它不是大而全的百科而是聚焦高频业务场景的映射表。例如当用户提到“AI 模型”词典会将其归类为inference_workload或training_workload并关联默认硬件偏好inference_workload → g5.xlarge,training_workload → p4d.24xlarge“快”被拆解为三个维度startup_time 60s对应 Spot 实例或 Serverless、compute_latency 10ms对应 GPU 实例、network_latency 50ms对应同 Region 部署“别太贵”被映射为预算等级budget_tier: medium对应月均 $2000-$5000并关联成本优化策略如“强制启用 Savings Plans”、“禁用按需实例”。其次我们设计了分阶段提示Multi-step Prompting。模型不是一次性输出最终 JSON而是分三步走意图识别先判断用户输入属于哪类需求Web App / Data Pipeline / AI Inference / Legacy Migration若置信度 0.85则触发澄清问题如“请问这个 AI 环境主要用于模型训练还是在线推理”约束提取在确认意图后专门抽取显性约束如“要 3 个副本”、“必须用 Aurora”和隐性约束如“金融行业”→ 自动添加compliance_tags: [PCI-DSS, SOC2]JSON 生成最后才生成严格遵循 Schema 的输出Schema 定义如下精简版{ service_type: string, scale_expectation: { qps: number, concurrent_users: number, data_volume_gb_per_day: number }, availability_requirement: string, // single_az, multi_az, global_active security_class: string, // public_facing, internal_only, pci_dss compliance_tags: [string], budget_tier: string, // low, medium, high preferred_region: string }提示这个 Schema 不是固定死的。我们每周会从历史交付日志中提取“用户追问高频问题”动态扩充 Schema 字段。例如上周发现 23% 的需求中用户会追问“备份策略”我们就新增了backup_policy对象包含retention_days,point_in_time_recovery_enabled等子字段。这种“数据驱动的 Schema 演进”让智能体越来越懂业务语言。3.2 架构设计智能体如何让“画图”不再拍脑袋架构设计环节最容易沦为“PPT 架构师”——画一堆漂亮的框但落地时发现 A 服务不支持 B 协议C 组件在 D Region 没有可用区。我们的解法是把架构决策变成一场“能力匹配游戏”。我们维护了一个名为cloud-capability-kb.json的知识库它不是静态文档而是一个可查询的、带版本号的 JSON 数据库。每条记录代表一个“云能力单元”结构如下{ id: aws-eks-autoscaling-v2, provider: aws, service: eks, feature: node_group_autoscaling, supported_regions: [us-east-1, us-west-2, ap-northeast-1], min_provider_version: 4.65.0, recommended_module: terraform-aws-eks, module_version_range: 22.0.0, 24.0.0, known_limitations: [不支持混合实例策略与 Fargate 同时启用], last_verified_at: 2024-05-10T08:22:15Z }当架构设计智能体收到需求如{service_type: ai_inference, scale_expectation: {qps: 5000}}它会执行以下步骤能力检索查询知识库找出所有匹配service_typeai_inference且qps 5000的能力单元得到候选列表如aws-eks-autoscaling-v2,aws-ecs-fargate-spot-v3,aws-lambda-provisioned-concurrency-v1约束过滤根据需求中的availability_requirement和security_class剔除不满足的选项如security_classpci_dss会过滤掉所有未通过 PCI 认证的 Fargate 配置成本排序调用云厂商的 Pricing Calculator API我们封装了 AWS/Azure/GCP 的统一接口计算每个候选方案的预估月成本按budget_tier过滤medium预算只保留 $1500-$4000 区间拓扑生成选定最优方案后不是手动画图而是调用预置的 Mermaid 模板引擎。例如选中aws-eks-autoscaling-v2就加载eks-inference-topology.mmd模板将占位符{{cluster_name}},{{node_count}}替换为实际值输出标准 Mermaid 代码。注意我们严禁智能体“自由发挥”画图。所有拓扑图必须源自经过验证的模板库模板由 SRE 团队每月评审更新。曾有一次模型试图“优化”一个 Kafka 集群图把 ZooKeeper 组件删掉了认为它过时结果生成的图无法部署。从此我们立下铁律架构图 模板 参数绝不允许模型修改拓扑结构本身。3.3 代码生成智能体为什么“抄作业”比“写作文”更可靠这是最受质疑的一环“让模型写代码不是更酷吗” 我的答案是在 IaC 领域“酷”等于事故。我们坚持“代码生成 高精度模板装配”原因有三可审计性每一行生成的代码都能追溯到模板库中的具体 commit ID 和行号。当安全团队要求审查“S3 存储桶的加密配置”我们直接给出templates/aws/s3-bucket-encrypted.tfv3.2.1#L45-L52而不是“模型根据上下文生成的”。可测试性模板库中的每个.tf文件都配有对应的test.tfvars和expected-output.json每天凌晨自动跑terraform test。新模板入库前必须通过 100% 的测试用例。可演进性当 AWS 发布新功能如 EKS 的 Bottlerocket 支持我们不是让模型去学而是 SRE 工程师写好新模板更新知识库智能体下次调用时自动切换。我们的模板库采用“分层继承”设计基础层base定义通用变量、Provider 配置、Backend 设置服务层service如eks-cluster,rds-postgresql,alb-application每个模板专注一个服务通过count和for_each支持多实例场景层scenario如ai-inference-stack,payment-gateway-stack它不写新资源而是组合服务层模板并注入场景特有逻辑如支付网关模板会自动添加 PCI 合规检查模块。代码生成智能体的工作流极其简单接收架构设计输出含recommended_module和module_version_range查询模板库找到匹配的场景模板如ai-inference-stack解析需求 JSON将scale_expectation.qps映射为eks_node_group_desired_capacity将compliance_tags注入所有资源的tags块执行terraform fmt格式化生成最终.tf文件。实操心得我们给每个模板配了一个README.md里面明确写着“适用场景”、“不适用场景”、“已知问题”。例如rds-postgresql模板的 README 第一行就写“⚠️ 本模板不适用于需要跨区域读副本的场景请改用rds-global-database模板”。这种“诚实的文档”比任何模型幻觉都管用。3.4 执行与验证智能体为什么“执行”才是最危险的环节很多人以为生成代码就结束了其实真正的风险在执行层。我们统计过过去一年中 73% 的 IaC 生产事故发生在terraform apply阶段而非代码生成阶段。原因很简单代码是静态的环境是动态的。terraform apply可能遭遇云 API 限流、资源配额不足、依赖服务临时不可用、甚至 Terraform Provider 本身的 bug。我们的执行与验证智能体不是简单地system(terraform apply)而是一个带七重防护的“手术台”沙箱预检Sandbox Pre-check在真实环境执行前先启动一个隔离的 Docker 容器挂载生成的代码和terraform.tfstate运行terraform init -backendfalse和terraform validate。这一步能捕获 90% 的语法错误和变量缺失。Plan 安全校验Plan Security Scanterraform plan -outplan.binary生成二进制计划后我们不直接执行而是用terraform show -json plan.binary解析为 JSON然后逐条检查是否有create操作涉及aws_iam_role且assume_role_policy包含Principal: {Service: ec2.amazonaws.com}这可能是提权风险是否有aws_s3_bucket的acl字段值为public-read所有aws_security_group_rule的cidr_blocks是否包含0.0.0.0/0。这些检查由 Checkov 的自定义规则驱动失败则熔断。资源拓扑比对Topology Diff将 Plan JSON 中的资源列表与需求 JSON 中的scale_expectation做数学比对。例如需求要求qps: 5000而 Plan 显示只创建了 2 个aws_eks_node_group每个desired_capacity: 2则总容量为 4低于需求自动拒绝执行。终态健康检查Post-deploy Health Checkapply成功后不立即返回“成功”而是启动一个轮询任务对 EKS 集群调用aws eks describe-cluster --name xxx检查status ACTIVE对 RDS 实例调用aws rds describe-db-instances --db-instance-identifier xxx检查DBInstanceStatus available且StorageEncrypted true对 ALB调用aws elbv2 describe-load-balancers --names xxx检查State.Code active。所有检查通过才标记交付完成。关键技巧我们给每个健康检查设置了“容忍窗口”。例如EKS 集群从CREATING到ACTIVE平均耗时 12 分钟但我们设置的超时是 25 分钟并且每 90 秒检查一次。这样既给了云平台足够时间又不会无限等待。这个窗口值不是拍脑袋而是从过去 30 天的交付日志中用 P95 分位数计算出来的。4. 实操过程与核心环节实现4.1 环境准备从零搭建四智能体流水线整个系统基于 AWS 云原生服务构建但所有组件均可替换为其他云厂商对应服务如 Azure 的 Logic Apps 替代 Step FunctionsGCP 的 Cloud Run 替代 ECS。以下是核心组件清单与部署逻辑组件选型理由关键配置AgentCore Runtime使用 ECS Fargate免运维、弹性伸缩CPU: 2 vCPU, Memory: 4GB, Log Group:/ecs/agentcoreRedis Stream作为智能体间消息总线低延迟、高吞吐Engine: Redis 7.0, Replicas: 1, Encryption at rest: enabledDynamoDB State Table存储任务状态支持高并发读写Billing mode: On-demand, TTL attribute:expires_atTerraform Backend使用 S3 DynamoDB Lock保证状态一致性S3 Bucket:iac-state-store-prod, DynamoDB Table:terraform-state-lockPricing Calculator API封装 AWS/Azure/GCP 定价接口统一调用部署在 Lambda冷启动优化预留并发 5内存 1024MB部署流程分四步全部通过 Terraform 代码化基础设施层用根模块创建 VPC、Subnet、Security Group、IAM Roles含AgentCoreExecutionRole权限最小化仅允许调用lambda:InvokeFunction,dynamodb:GetItem,s3:GetObject数据层创建 Redis Cluster 和 DynamoDB Table其中 DynamoDB 的Partition Key设为task_idSort Key设为step_id如req-parse,arch-design,code-gen方便按任务查询全流程状态运行时层部署 ECS Cluster 和 Task Definition其中containerDefinitions指定 AgentCore 的 Docker 镜像我们基于官方镜像构建预装了aws-cli,terraform,tflint,checkov智能体层为每个智能体创建独立的 ECS Service配置不同的ENTRYPOINT如需求理解智能体的 ENTRYPOINT 是python /app/interpreter.py并通过 ECS Service Discovery 实现服务发现。提示我们为每个智能体 Service 配置了独立的 Auto Scaling Policy但策略依据不是 CPU而是 Redis Stream 的 Pending Entries 数量。当req-interpreter-stream的 pending count 10就扩容 interpreter Service 的 Task Count。这种“事件驱动的弹性”比传统指标更精准。4.2 智能体开发以“需求理解智能体”为例的完整代码结构我们用 Python 开发所有智能体框架基于 FastAPI轻量、异步、OpenAPI 原生支持。以需求理解智能体interpreter.py为例其核心结构如下# interpreter.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel import re import json from typing import Dict, Any app FastAPI(titleRequirement Interpreter) class RequirementInput(BaseModel): raw_text: str session_id: str # 用于关联用户会话 class RequirementOutput(BaseModel): service_type: str scale_expectation: Dict[str, float] availability_requirement: str security_class: str compliance_tags: list budget_tier: str preferred_region: str # 加载语义词典从 S3 下载缓存到内存 SEMANTIC_DICT load_from_s3(s3://iac-kb/semantic-dict.json) app.post(/interpret, response_modelRequirementOutput) async def interpret_requirement(input: RequirementInput): try: # 步骤1输入清洗 cleaned_text clean_input(input.raw_text) # 步骤2意图识别调用微调的小模型非大模型 intent predict_intent(cleaned_text) # 模型输出web_app, ai_inference, data_pipeline... if not intent or confidence_score 0.85: raise HTTPException(status_code400, detailIntent unclear, please clarify) # 步骤3约束提取正则 词典匹配 constraints extract_constraints(cleaned_text, SEMANTIC_DICT) # 步骤4JSON 生成调用 LLM但 prompt 严格限定为 JSON 输出 result_json generate_structured_output(cleaned_text, constraints, intent) # 步骤5Schema 校验 validated RequirementOutput.parse_obj(result_json) # 步骤6存入 DynamoDB 状态表 save_to_dynamodb(input.session_id, req-parse, validated.dict()) return validated except Exception as e: save_to_dynamodb(input.session_id, req-parse, {error: str(e)}) raise HTTPException(status_code500, detailfInterpretation failed: {e}) def clean_input(text: str) - str: # 移除特殊字符标准化空格转换为小写 return re.sub(r[^\w\s\-], , text).strip().lower() def predict_intent(text: str) - str: # 调用微调的 DistilBERT 模型12MB部署在 SageMaker Real-time Endpoint # 输入text输出intent_label, confidence_score pass def extract_constraints(text: str, dict_data: dict) - dict: # 遍历语义词典匹配关键词填充约束 constraints {} for keyword, mapping in dict_data.items(): if keyword in text: constraints.update(mapping) return constraints def generate_structured_output(text: str, constraints: dict, intent: str) - dict: # 调用 Bedrock 的 Claude 3 Haiku低成本、高响应速度 # Prompt 模板 # # 你是一个专业的云架构需求分析师。请将以下用户需求严格按 JSON Schema 输出不要任何额外文字。 # 用户需求{text} # 已识别意图{intent} # 已提取约束{constraints} # JSON Schema: {RequirementOutput.schema_json()} # pass关键细节我们没有用一个大模型干所有事而是“小模型做意图识别 大模型做 JSON 生成”。因为意图识别是分类任务DistilBERT 就够了准确率 92%延迟 200ms而 JSON 生成需要强推理交给 Claude 3 Haiku它在 1000 token 内的 JSON 生成准确率高达 98.7%。这种“大小模型协同”比单一大模型更稳、更快、更便宜。4.3 MCP 控制层实现三道防线的代码级落地MCP 的三道防线全部通过中间件Middleware方式注入到 FastAPI 应用中与业务逻辑解耦# mcp_middleware.py from fastapi import Request, Response, HTTPException from starlette.middleware.base import BaseHTTPMiddleware import hashlib import time class InputSanitizerMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): # 读取请求体 body await request.body() text body.decode(utf-8) # 正则过滤高危词 dangerous_patterns [rrm\s-rf, rdestroy, rforce-unlock, r--auto-approve] for pattern in dangerous_patterns: if re.search(pattern, text, re.IGNORECASE): raise HTTPException(status_code400, detailInput contains prohibited command) # 业务术语映射 term_mapping { 删掉旧库: set deletion_protectionfalse for prod-db-old, 关掉监控: disable CloudWatch alarms for resource X } for src, dst in term_mapping.items(): text text.replace(src, dst) # 重写请求体 request._body text.encode(utf-8) return await call_next(request) class OutputValidatorMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): response await call_next(request) # 拦截响应体假设是 JSON if response.headers.get(content-type) application/json: body b async for chunk in response.body_iterator: body chunk try: data json.loads(body.decode(utf-8)) # 如果是代码生成响应执行安全扫描 if generated_code in data: scan_result run_checkov_scan(data[generated_code]) if not scan_result[passed]: raise HTTPException(status_code422, detailfSecurity scan failed: {scan_result[issues]}) # 添加 MCP 签名 signature hashlib.sha256(body).hexdigest()[:16] data[mcp_signature] fsha256:{signature} response.body_iterator iter([json.dumps(data, ensure_asciiFalse).encode(utf-8)]) except Exception as e: raise HTTPException(status_code500, detailfOutput validation error: {e}) return response class ExecutionCircuitBreaker: def __init__(self): self.credit_pool {} def check_credit(self,