AWS Bedrock提示词管理与流程化实战

如果你的团队正在构建 LLM 应用，那么提示词很可能散落在应用代码中，以字符串字面量的形式存在。改动一个词就意味着一次代码变更、一次 review、一次重新部署——而且没有人能说清楚生产环境中实际运行的是哪个版本的提示词。随着 LLM 应用从原型走向生产系统，提示词蔓延成了一个真实的运维问题。

在本手把手教程中，我们将通过两个 Amazon Bedrock 功能——Prompt Management 和 Prompt Flows——来解决这个问题，构建一个客户服务流水线，对客户咨询进行分类、路由，并使用合适的模型回答。全部通过 AWS CLI 操作，每个命令都有记录。

更喜欢视频？整个手把手教程也有 YouTube 版：

客户咨询
      │
      ▼
 [分类器提示词]  ← Amazon Nova Micro（便宜，temperature 0）
      │
      ▼
   [路由器]  ── TECH ──▶ [技术回答提示词]    ← Claude（准确）
      │
      └── 其他一切 ──▶ [通用回答提示词] ← Nova Micro（便宜）

注意成本设计：分类和通用回答运行在 Amazon Nova Micro 上（快速且廉价），而只有真正技术性的问题才会触达 Claude。将廉价流量从最贵的模型上分流，是生产级 LLM 系统中最实用的成本优化手段之一。

什么是 Prompt Management？

Prompt Management 将提示词变成了一等 AWS 资源。提示词不再是代码中的一个字符串，而是一个拥有自己 ARN 的对象，携带运行所需的一切：

• 带变量的模板——例如 {{question}}——在调用时填充
• 绑定的模型——每个提示词都知道自己在哪个模型上运行（modelId）
• 推理配置——temperature、max tokens 等，随提示词一起存储
• 版本——一个可编辑的 DRAFT 加上不可变的编号快照

这些特性的影响比技术细节更重要。你的应用代码不再包含任何提示词文本——它只引用一个提示词 ARN。提示词工程师可以迭代 DRAFT 而无需触碰应用代码，而生产环境继续调用固定的版本。为提示词更换底层模型是配置变更，而非代码变更。

什么是 Prompt Flows？

Prompt Flows 是 Bedrock 的无服务器编排层，用于多步骤 LLM 流水线。Flow 是一个由节点组成的图——Input、Prompt、Condition、Output 等——通过数据和条件边连接。Flow 引擎帮你执行这个图：无需 Lambda 粘合代码，无需维护 Step Functions 状态机。

Flow 同样拥有与提示词相同的生命周期管理：你可以对 Flow 进行版本控制，并通过别名暴露（例如一个指向版本 1 的 prod 别名），这样你可以在不重新部署调用方的情况下重新连接流水线。

两个功能结合，给你团队通常需要自己手动实现的东西：一个带版本控制的提示词注册表，外加一个带路由功能的托管执行引擎——两者都可通过标准 AWS API 和 IAM 调用。

前置条件

• AWS CLI v2 已配置，能够访问 us-east-1 区域的 Bedrock
• 在 Bedrock 控制台中已为 Claude 和 Nova 模型启用模型访问
• 已安装 jq 和 python3（带有 boto3）

步骤 1：Prompt Management

设置环境变量

Bedrock 模型更新频繁——请查看控制台并使用最新版本。

export CLAUDE_MODEL="us.anthropic.claude-haiku-4-5-20251001-v1:0"
export NOVA_MODEL="us.amazon.nova-micro-v1:0"
export REGION=us-east-1

首先创建分类器。Temperature 设为 0，token 上限设为 10：我们需要一个确定性的标签，不需要其他内容。

CLASSIFIER_ID=$(aws bedrock-agent create-prompt --region $REGION \
  --name customer-classifier \
  --description "Classify customer inquiries as TECH or GENERAL" \
  --default-variant v1 \
  --variants '[{
    "name":"v1",
    "templateType":"TEXT",
    "templateConfiguration":{"text":{
      "text":"Classify the following customer inquiry into exactly one of the following. Respond with the label only and nothing else.\n\n- TECH: technical product issues, configuration, errors, how to use the product\n- GENERAL: pricing, contracts, business hours, and other general questions\n\nInquiry: {{question}}\n\nLabel:",
      "inputVariables":[{"name":"question"}]
    }},
    "modelId":"'"$NOVA_MODEL"'",
    "inferenceConfiguration":{"text":{"temperature":0.0,"maxTokens":10}}
  }]' \
  --query 'id' --output text)

echo "CLASSIFIER_ID=$CLASSIFIER_ID"

接下来是技术回答器——这个运行在 Claude 上，使用较低的 temperature 以保证准确性：

TECH_ID=$(aws bedrock-agent create-prompt --region $REGION \
  --name customer-tech-answer \
  --description "Answer technical inquiries with Claude" \
  --default-variant v1 \
  --variants '[{
    "name":"v1",
    "templateType":"TEXT",
    "templateConfiguration":{"text":{
      "text":"You are a technical support representative for our products. Answer the following technical inquiry concisely and accurately.\n\nInquiry: {{question}}\n\nAnswer:",
      "inputVariables":[{"name":"question"}]
    }},
    "modelId":"'"$CLAUDE_MODEL"'",
    "inferenceConfiguration":{"text":{"temperature":0.3,"maxTokens":400}}
  }]' \
  --query 'id' --output text)

echo "TECH_ID=$TECH_ID"

然后是通用回答器，回到 Nova Micro：

GENERAL_ID=$(aws bedrock-agent create-prompt --region $REGION \
  --name customer-general-answer \
  --description "Answer general inquiries with Nova" \
  --default-variant v1 \
  --variants '[{
    "name":"v1",
    "templateType":"TEXT",
    "templateConfiguration":{"text":{
      "text":"You are a customer support representative for our company. Answer the following general inquiry politely and helpfully.\n\nInquiry: {{question}}\n\nAnswer:",
      "inputVariables":[{"name":"question"}]
    }},
    "modelId":"'"$NOVA_MODEL"'",
    "inferenceConfiguration":{"text":{"temperature":0.5,"maxTokens":400}}
  }]' \
  --query 'id' --output text)

echo "GENERAL_ID=$GENERAL_ID"

三个提示词，三个独立的生命周期，两种不同的模型——还没写一行应用代码。

直接调用托管提示词

这里有个让大多数人惊讶的细节：Converse API 接受一个提示词 ARN 作为模型 ID。提示词自带模型和推理设置；你只需提供变量。

ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
CLASSIFIER_ARN="arn:aws:bedrock:$REGION:$ACCOUNT_ID:prompt/$CLASSIFIER_ID"

aws bedrock-runtime converse --region $REGION \
  --model-id "$CLASSIFIER_ARN" \
  --prompt-variables '{"question":{"text":"I cannot log in. I want to reset my password."}}' \
  | jq -r '.output.message.content[0].text'

预期输出：TECH

aws bedrock-runtime converse --region $REGION \
  --model-id "$CLASSIFIER_ARN" \
  --prompt-variables '{"question":{"text":"How much is the monthly fee?"}}' \
  | jq -r '.output.message.content[0].text'

预期输出：GENERAL

版本控制：DRAFT / 固定版本工作流

每个提示词都有一个可编辑的 DRAFT。create-prompt-version 将当前 DRAFT 冻结为一个不可变的编号版本：

CLASSIFIER_V1=$(aws bedrock-agent create-prompt-version --region $REGION \
  --prompt-identifier $CLASSIFIER_ID \
  --description "Production version" \
  --query version --output text)

echo "CLASSIFIER_V1=$CLASSIFIER_V1"   # -> 1

TECH_V1=$(aws bedrock-agent create-prompt-version --region $REGION \
  --prompt-identifier $TECH_ID --query version --output text)
GENERAL_V1=$(aws bedrock-agent create-prompt-version --region $REGION \
  --prompt-identifier $GENERAL_ID --query version --output text)
echo "TECH_V1=$TECH_V1  GENERAL_V1=$GENERAL_V1"

ARN 约定完成了剩余工作。追加 :1 获取固定的生产版本；省略后缀则访问 DRAFT：

# 生产环境：固定到版本 1
aws bedrock-runtime converse --region $REGION \
  --model-id "$CLASSIFIER_ARN:$CLASSIFIER_V1" \
  --prompt-variables '{"question":{"text":"I want to reset my password"}}' \
  | jq -r '.output.message.content[0].text'

# 测试：DRAFT，包含任何正在进行的编辑
aws bedrock-runtime converse --region $REGION \
  --model-id "$CLASSIFIER_ARN" \
  --prompt-variables '{"question":{"text":"I want to reset my password"}}' \
  | jq -r '.output.message.content[0].text'

我们来证明隔离性。重写 DRAFT，添加第三个标签 BILLING：

现在用同一个账单问题问两次：

# DRAFT — 拾取新行为
aws bedrock-runtime converse --region $REGION \
  --model-id "$CLASSIFIER_ARN" \
  --prompt-variables '{"question":{"text":"How much is the monthly fee?"}}' \
  | jq -r '.output.message.content[0].text'
# -> BILLING

# Version 1 — 生产环境未受影响
aws bedrock-runtime converse --region $REGION \
  --model-id "$CLASSIFIER_ARN:1" \
  --prompt-variables '{"question":{"text":"How much is the monthly fee?"}}' \
  | jq -r '.output.message.content[0].text'
# -> GENERAL

同一个 prompt 资源，两种行为，对生产零风险。这就是 Prompt Management 存在的意义。

现在我们将三个 prompt 连接到一条流水线中。

创建执行角色

Flow 在自己的 IAM 角色下运行，该角色需要权限来调用模型和读取 prompt：

cat > /tmp/flow-trust.json << 'EOF'
{
  "Version":"2012-10-17",
  "Statement":[{"Effect":"Allow","Principal":{"Service":"bedrock.amazonaws.com"},"Action":"sts:AssumeRole"}]
}
EOF

FLOW_ROLE_ARN=$(aws iam create-role \
  --role-name bedrock-flow-role \
  --assume-role-policy-document file:///tmp/flow-trust.json \
  --query 'Role.Arn' --output text)

cat > /tmp/flow-policy.json << 'EOF'
{
  "Version":"2012-10-17",
  "Statement":[
    {"Effect":"Allow","Action":["bedrock:InvokeModel","bedrock:Converse","bedrock:GetPrompt","bedrock:RenderPrompt","bedrock:GetInferenceProfile"],"Resource":"*"}
  ]
}
EOF

aws iam put-role-policy \
  --role-name bedrock-flow-role \
  --policy-name bedrock-flow-inline \
  --policy-document file:///tmp/flow-policy.json

echo "FLOW_ROLE_ARN=$FLOW_ROLE_ARN"
sleep 10   # 等待角色传播

（生产环境下，请将 Resource 范围缩小到具体的 prompt 和模型 ARN。）

将 Flow 定义为 JSON

Flow 定义由两部分组成：节点（盒子）和连接（箭头）。我们的 Flow 有一个 Input 节点、三个引用了 pin 到生产版本的 prompt 节点、一个用于路由的 Condition 节点和两个 Output 节点：

CLASSIFIER_PROD_ARN="$CLASSIFIER_ARN:$CLASSIFIER_V1"
TECH_PROD_ARN="arn:aws:bedrock:$REGION:$ACCOUNT_ID:prompt/$TECH_ID:$TECH_V1"
GENERAL_PROD_ARN="arn:aws:bedrock:$REGION:$ACCOUNT_ID:prompt/$GENERAL_ID:$GENERAL_V1"

jq -n \
  --arg cls "$CLASSIFIER_PROD_ARN" \
  --arg tech "$TECH_PROD_ARN" \
  --arg gen "$GENERAL_PROD_ARN" \
'{
  nodes: [
    { name:"FlowInputNode", type:"Input",
      configuration:{input:{}},
      outputs:[{name:"document", type:"String"}] },

{ name:"Classifier", type:"Prompt",
      configuration:{prompt:{sourceConfiguration:{resource:{promptArn:$cls}}}},
      inputs:[{name:"question", type:"String", expression:"$.data"}],
      outputs:[{name:"modelCompletion", type:"String"}] },

{ name:"Router", type:"Condition",
      configuration:{condition:{conditions:[
        {name:"isTech", expression:"classification == \"TECH\""},
        {name:"default"}
      ]}},
      inputs:[{name:"classification", type:"String", expression:"$.data"}] },

{ name:"TechAnswer", type:"Prompt",
      configuration:{prompt:{sourceConfiguration:{resource:{promptArn:$tech}}}},
      inputs:[{name:"question", type:"String", expression:"$.data"}],
      outputs:[{name:"modelCompletion", type:"String"}] },

{ name:"GeneralAnswer", type:"Prompt",
      configuration:{prompt:{sourceConfiguration:{resource:{promptArn:$gen}}}},
      inputs:[{name:"question", type:"String", expression:"$.data"}],
      outputs:[{name:"modelCompletion", type:"String"}] },

{ name:"TechOutputNode", type:"Output",
      configuration:{output:{}},
      inputs:[{name:"document", type:"String", expression:"$.data"}] },

{ name:"GeneralOutputNode", type:"Output",
      configuration:{output:{}},
      inputs:[{name:"document", type:"String", expression:"$.data"}] }
  ],
  connections: [
    {name:"i2c", source:"FlowInputNode", target:"Classifier",         type:"Data",
      configuration:{data:{sourceOutput:"document", targetInput:"question"}}},
    {name:"i2t", source:"FlowInputNode", target:"TechAnswer",         type:"Data",
      configuration:{data:{sourceOutput:"document", targetInput:"question"}}},
    {name:"i2g", source:"FlowInputNode", target:"GeneralAnswer",      type:"Data",
      configuration:{data:{sourceOutput:"document", targetInput:"question"}}},
    {name:"c2r", source:"Classifier",    target:"Router",             type:"Data",
      configuration:{data:{sourceOutput:"modelCompletion", targetInput:"classification"}}},
    {name:"r2t", source:"Router",        target:"TechAnswer",         type:"Conditional",
      configuration:{conditional:{condition:"isTech"}}},
    {name:"r2g", source:"Router",        target:"GeneralAnswer",      type:"Conditional",
      configuration:{conditional:{condition:"default"}}},
    {name:"t2o", source:"TechAnswer",    target:"TechOutputNode",     type:"Data",
      configuration:{data:{sourceOutput:"modelCompletion", targetInput:"document"}}},
    {name:"g2o", source:"GeneralAnswer", target:"GeneralOutputNode",  type:"Data",
      configuration:{data:{sourceOutput:"modelCompletion", targetInput:"document"}}}
  ]
}' > /tmp/flow-def.json

cat /tmp/flow-def.json | jq '.nodes[].name, .connections[].name'

有两个细节值得注意：

数据边 vs 条件边。 问题通过 Data 连接流向所有三个 prompt 节点，但 TechAnswer 和 GeneralAnswer 只有在来自 Router 的 Conditional 边触发时才执行。Condition 节点控制执行流程，而非数据本身。

Router 的表达式（classification == "TECH"）与分类器的原始输出进行比较——这正是我们强制分类器只输出标签的原因。

创建、准备和测试 Flow

FLOW_ID=$(aws bedrock-agent create-flow --region $REGION \
  --name customer-support-flow \
  --description "Classify -> route -> answer" \
  --execution-role-arn "$FLOW_ROLE_ARN" \
  --definition file:///tmp/flow-def.json \
  --query 'id' --output text)

echo "FLOW_ID=$FLOW_ID"

# 将 Flow 编译为可执行状态
aws bedrock-agent prepare-flow --region $REGION \
  --flow-identifier $FLOW_ID

invoke_flow 会流式返回事件，所以我们用 Python 来测试。TSTALIASID 是内置别名，始终指向 Flow 的工作草稿：

import boto3

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

def ask(question):
    resp = client.invoke_flow(
        flowIdentifier="<FLOW_ID>",
        flowAliasIdentifier="TSTALIASID",
        inputs=[{
            "nodeName": "FlowInputNode",
            "nodeOutputName": "document",
            "content": {"document": question}
        }]
    )
    for event in resp["responseStream"]:
        if "flowOutputEvent" in event:
            out = event["flowOutputEvent"]
            print(f"  [Output node: {out['nodeName']}]")
            print("  " + out["content"]["document"].replace("\n", "\n  "))

print("=== Technical question ===")
ask("The app crashes as soon as I launch it. What could be the cause?")

print()
print("=== General question ===")
ask("What are the support center's business hours?")

技术问题从 TechOutputNode 返回（由 Claude 回答），营业时间问题从 GeneralOutputNode 返回（由 Nova 回答）。路由生效了。

Flow 版本和别名

Flow 的版本管理与 prompt 完全一致——别名给调用方一个稳定的名称：

FLOW_V1=$(aws bedrock-agent create-flow-version --region $REGION \
  --flow-identifier $FLOW_ID \
  --description "Production release v1" \
  --query version --output text)

echo "FLOW_V1=$FLOW_V1"   # -> 1

PROD_ALIAS_ID=$(aws bedrock-agent create-flow-alias --region $REGION \
  --flow-identifier $FLOW_ID \
  --name prod \
  --routing-configuration '[{"flowVersion":"'"$FLOW_V1"'"}]' \
  --query id --output text)

echo "PROD_ALIAS_ID=$PROD_ALIAS_ID"

生产环境调用方使用 prod 别名，永远不变——即使你之后将别名重新指向版本 2：

import boto3
client = boto3.client("bedrock-agent-runtime", region_name="us-east-1")
resp = client.invoke_flow(
    flowIdentifier="<FLOW_ID>",
    flowAliasIdentifier="<PROD_ALIAS_ID>",
    inputs=[{
        "nodeName": "FlowInputNode",
        "nodeOutputName": "document",
        "content": {"document": "How do I get a receipt issued?"}
    }]
)
for event in resp["responseStream"]:
    if "flowOutputEvent" in event:
        out = event["flowOutputEvent"]
        print(f"[{out['nodeName']}]")
        print(out["content"]["document"])

删除子资源后再删除父资源：

# Flow（别名 -> 版本 -> flow）
aws bedrock-agent delete-flow-alias --region $REGION \
  --flow-identifier $FLOW_ID --alias-identifier $PROD_ALIAS_ID
aws bedrock-agent delete-flow-version --region $REGION \
  --flow-identifier $FLOW_ID --flow-version $FLOW_V1
aws bedrock-agent delete-flow --region $REGION \
  --flow-identifier $FLOW_ID --skip-resource-in-use-check

# Prompts
for P in $CLASSIFIER_ID $TECH_ID $GENERAL_ID; do
  aws bedrock-agent delete-prompt --region $REGION --prompt-identifier $P
done

# IAM
aws iam delete-role-policy --role-name bedrock-flow-role --policy-name bedrock-flow-inline
aws iam delete-role --role-name bedrock-flow-role

Prompt Management 和 Prompt Flows 不仅仅是便捷功能——它们将 prompt 和 LLM 流水线从应用代码中剥离出来，变成了版本化的、受 IAM 管控的 AWS 资源。DRAFT/version 分裂机制为 prompt 工程提供了一个安全的生产环境测试工作流，而 flow 别名则让你可以在一个稳定的端点背后重新构建流水线架构。如果你目前在 AWS 上运行 LLM 工作负载，我建议在自己的环境中验证这一模式。

Maruchin Tech — 12 项 AWS 认证 | 制造业与供应链的云与 AI（AWS / Google Cloud / Azure）| Udemy 讲师（学员 10 万+）

🎥 手把手操作视频版：https://youtu.be/iyK_xK3G-i0

📚 完整课程 — AWS Certified Generative AI Developer Professional（AIP-C01）考试备考：https://www.udemy.com/course/aws-certified-generative-ai-developer-professional-exam-prep/

👨‍🏫 全部课程（Udemy 个人主页）：https://www.udemy.com/user/maruchin-tech-2/

🎫 每月折扣券：https://www.youtube.com/@MaruchinTech-cloud/community