使用 AgentCore 与 Strands 构建：受治理的多 Agent 风险系统开发者工作坊

目标受众： 资深开发者、平台工程师、AI 治理工程师

时长： 2 小时

主要 AWS AI 服务： Amazon Bedrock AgentCore Runtime、AgentCore Memory 模式、AgentCore Policy 模式、AgentCore Observability 模式、AgentCore Evaluations 模式、Strands Agents

项目产出： 一个具备专业 Strands Agent、有界记忆、策略检查、结构化遥测、评估测试组件和运行时客户端的受治理多 Agent 运行时（Runtime）。

仅供教育用途之工程工作坊。本活动为软件架构演练，不构成财务建议。

工作坊摘要

本工作坊将教导开发者如何使用 AgentCore Runtime 与 Strands 专家 Agent 设计一个受治理的多 Agent 风险系统。参与者将实施协调编排、请求与响应的策略检查、有界记忆摘要、结构化可观测性事件，以及针对安全与阻挡场景的评估测试组件。最终成果是一个可重复使用的架构，为真实团队在 AWS 上实现证据优先的综合分析、可追溯的运营，以及更安全的多 Agent 协作工作流。

---

1. 开发者学习目标

开发者将学习如何：

● 将 AI 工作流分解为编排器（Orchestrator）与专家（Specialist）Agent。

● 建立具备领域特定工具的专业 Strands Agent。

● 构建 AgentCore Runtime 编排器进入点。

● 在模型调用前加入策略检查（Policy Checks）。

● 加入有界记忆摘要以确保安全连续性。

● 发送结构化可观测性事件。

● 针对安全与阻挡流程建立评估测试组件（Evaluation Fixtures）。

● 构建运行时客户端并测试多 Agent 行为。

---

2. 本工作坊构建的架构

Client (用戶端)
  └─ invoke_governed_runtime.py

AgentCore Runtime
  └─ orchestrator.py
       ├─ policy.py              # 預檢與回應檢查
       ├─ memory_store.py        # 有界摘要記憶
       ├─ observability.py       # 結構化遙測
       ├─ specialists.py         # 流動性 / 信用 / 外匯 / 主權專家 Agent
       └─ 評估測試組件

Strands Agents
  ├─ 流動性專家 Agent
  ├─ 信用專家 Agent
  ├─ 外匯專家 Agent
  ├─ 主權專家 Agent
  └─ 最終編排綜合 Agent

---

3. 2 小时实施议程

---

步骤 1 — 建立提示词契约

开发者执行动作

mkdir -p agentcore-governed-risk/prompts
cd agentcore-governed-risk
cat > prompts/orchestrator.md <<'EOF'
你是一個主權風險編排器。
不要獨自回答所有問題。將請求分解為專家任務。
使用流動性專家來處理資金壓力與現金壓力。
使用信用專家來處理利差擴大與違約風險重新定價。
使用外匯專家來處理貨幣錯配、避險背景與資本流動壓力。
使用主權專家來處理財政公信力、實質殖利率、政策分歧與債務重新定價。
將專家的證據合併為：證據、風險體制、避險考量、
確認訊號、失效觸發因素、未決問題和限制。
請勿提供自主交易指令或個人化財務建議。
EOF

建立 requirements.txt。

bedrock-agentcore
strands-agents
strands-agents-tools
boto3
pytest

商业逻辑

编排器提示词定义了多 Agent 工作流与最终响应结构。

程序逻辑

该提示词由编排器运行时加载，并控制最终的综合分析行为。

预期结果

代码库中包含一个定义清晰的编排器角色契约。

系统设计决策

● 每个角色的提示词契约： 多 Agent 系统需要清晰的角色边界。编排器提示词定义了委派与综合的职责，减少单一 Agent 试图处理所有事情的机率。

● 提示词中的输出纲要（Schema）： 必要的段落使最终答案更容易测试与显示。评估机制可以检查是否包含确认信号、失效触发因素和限制。

● 每个角色中的安全边界： 提示词中明确排除了自主交易与个性化财务建议。这不是唯一的控制手段，但在策略代码检查输出之前，它能引导模型的行为。

---

步骤 2 — 构建专家 Agent

开发者执行动作

建立 specialists.py。

from dataclasses import dataclass
from strands import Agent, tool
from strands.models import BedrockModel

MODEL_ID = "amazon.nova-pro-v1:0"

@dataclass
class SpecialistResult:
    name: str
    evidence: str
    confidence: str
    missing_data: str

@tool
def bps_change(current: float, previous: float) -> str:
    """計算基點（bps）的變化。"""
    return f"變化: {(current - previous) * 100:.1f} bps"

@tool
def hedge_amount(exposure: float, hedge_ratio: float) -> str:
    """根據曝險和避險比率計算避險金額。"""
    return f"避險金額: {exposure * hedge_ratio:,.2f}"

@tool
def liquidity_buffer(required_outflow: float, buffer_ratio: float) -> str:
    """計算流動性緩衝需求。"""
    return f"所需的流動性緩衝: {required_outflow * buffer_ratio:,.2f}"

def build_specialist(name: str, focus: str, tools=None) -> Agent:
    return Agent(
        model=BedrockModel(model_id=MODEL_ID, temperature=0.2, max_tokens=2000),
        tools=tools or [],
        system_prompt=(
            f"你是 {name} 專家。請僅專注於 {focus}。"
            "傳回證據、信心度、缺失資料和限制。"
            "請勿提供投資建議。"
        ),
    )

liquidity_agent = build_specialist(
    "liquidity",
    "資金流動性、回購壓力、現金偏好和市場深度",
    [bps_change, liquidity_buffer],
)

credit_agent = build_specialist(
    "credit",
    "信用利差擴大、評等調降壓力、違約風險重新定價和流動性溢價",
    [bps_change],
)

fx_agent = build_specialist(
    "fx",
    "貨幣錯配、資本流動、人民幣壓力、基差換匯和避險背景",
    [hedge_amount],
)

sovereign_agent = build_specialist(
    "sovereign",
    "實質殖利率、財政公信力、政策分歧、央行反應和債務重新定價",
    [bps_change],
)

def run_specialist(agent: Agent, name: str, task: str) -> SpecialistResult:
    response = agent(task).message["content"][0]["text"]
    return SpecialistResult(
        name=name,
        evidence=response,
        confidence="medium",
        missing_data="請參閱專家證據以獲取所需的缺失資料。",
    )

商业逻辑

每个专家仅专注于风险的一个维度，并且只接收该维度所需的工具。

程序逻辑

该文件定义了一个数据类别（Dataclass）结果契约、可重复使用的工具函数、一个专家工厂、四个 Agent 以及一个执行辅助程序。

预期结果

编排器可以导入专家并分派专注的任务给他们。

系统设计决策

● 缩小专家范围： 精简的提示词可以减少认知负载，并使 Agent 的输出更具焦。每个专家随后可以根据领域特定的标准进行独立评估。

● 最小权限工具分配： 流动性专家获得缓冲与基点工具，外汇专家获得避险规模计算，而专注于利差的 Agent 则获得基点计算工具。这减少了工具误用的机会并支持策略执行。

● 数据类别结果契约： 类型安全的契约有助于编排、遥测和评估。即使 Agent 响应本身是自然语言，编排器也能收到一致的字段。

---

步骤 3 — 新增策略层

开发者执行动作

建立 policy.py。

BLOCKED_REQUEST_PATTERNS = [
    "執行交易",
    "下單",
    "繞過審批",
    "保證獲利",
    "隱瞞損失",
    "規避控制",
    "忽略風險限制",
]

REQUIRED_RESPONSE_TERMS = ["確認", "失效", "限制"]

def validate_request(prompt: str) -> dict:
    text = prompt.lower()
    for pattern in BLOCKED_REQUEST_PATTERNS:
        if pattern in text:
            return {"allowed": False, "reason": f"遭阻擋的不支援請求模式: {pattern}"}
    return {"allowed": True, "reason": "允許用於教育分析。"}

def validate_response(response: str) -> dict:
    text = response.lower()
    checks = {term: term in text for term in REQUIRED_RESPONSE_TERMS}
    checks["advice_boundary"] = "非投資建議" in text or "非財務建議" in text
    return {"passed": all(checks.values()), "checks": checks}

商业逻辑

策略层会阻挡不受支持的自主行动请求，并验证最终响应的结构。

程序逻辑

validate_request() 在模型调用前执行。validate_response() 可以在综合分析后或在评估脚本中执行。

预期结果

不安全的执行类提示词会在专家 Agent 执行前被阻挡。

系统设计决策

● 策略先于成本： 在模型调用前阻挡不受支持的请求可以节省成本并降低风险。这也为明显无效的提示词提供了确定性的行为。

● 输入与输出控制： 输入策略阻挡不良请求。输出策略验证必要的治理段落。两者皆不可或缺，因为单靠提示词无法保证合规的输出。

● 便于工作坊理解的简易规则： 字符串匹配在手把手实验室中非常易于理解。实际生产环境的实施可以用 AgentCore Policy、分类器、身份识别感知规则和工具调用验证来取代。

---

步骤 4 — 新增有界记忆

开发者执行动作

建立 memory_store.py。

import json
from pathlib import Path
from datetime import datetime, timezone

MEMORY_FILE = Path("memory_state.json")
MAX_ITEMS_PER_USER = 10
MAX_ITEMS_IN_PROMPT = 3

def _read_all() -> dict:
    if not MEMORY_FILE.exists():
        return {}
    return json.loads(MEMORY_FILE.read_text(encoding="utf-8"))

def _write_all(data: dict):
    MEMORY_FILE.write_text(json.dumps(data, indent=2), encoding="utf-8")

def load_memory(user_id: str) -> str:
    data = _read_all()
    items = data.get(user_id, [])[-MAX_ITEMS_IN_PROMPT:]
    if not items:
        return "無先前安全記憶。"
    return "\n".join(item["summary"] for item in items)

def save_memory_summary(user_id: str, prompt: str, response: str):
    data = _read_all()
    item = {
        "created_at": datetime.now(timezone.utc).isoformat(),
        "summary": "先前的活頁工作流請求了主權風險分析。僅保留工作流上下文，不保留個人財務指令。",
        "prompt_chars": len(prompt),
        "response_chars": len(response),
    }
    data.setdefault(user_id, []).append(item)
    data[user_id] = data[user_id][-MAX_ITEMS_PER_USER:]
    _write_all(data)

商业逻辑

记忆功能保留了安全的工作流连续性，同时避免存储原始的对话记录。

程序逻辑

该模块读取/写入 JSON，仅返回最新的安全摘要，并限制保留数量。

预期结果

来自同一用户的重复请求将包含有界的先前工作流上下文。

系统设计决策

● 摘要记忆而非对话记录记忆： 原始提示词可能包含敏感或不相关的内容。存储安全的摘要可以减少风险暴露，并防止旧的模型文字被盲目重复使用。

● 有界回溯（Bounded Recall）： 提示词中仅注入三个摘要。这可以控制提示词的大小并降低过期上下文的风险。持久化存储也只保留有限的历史记录。

● 可替换的存储抽象： 本地文件实施旨在教学基础概念。生产系统可以直接替换为托管的 AgentCore Memory，而无需重写编排器逻辑。

---

步骤 5 — 新增结构化可观测性

开发者执行动作

建立 observability.py。

import json
import time
from datetime import datetime, timezone

START = time.time()

def emit_event(event_type: str, request_id: str, attributes: dict):
    event = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "elapsed_ms": int((time.time() - START) * 1000),
        "event_type": event_type,
        "request_id": request_id,
        "attributes": attributes,
    }
    print(json.dumps(event, ensure_ascii=False))

商业逻辑

遥测记录包含策略决策、编排启动、专家完成以及最终综合完成等事件。

程序逻辑

单一辅助程序，负责列印包含时间戳记、经过毫秒数、事件类型、请求 ID 和属性的结构化 JSON 事件。

预期结果

运行时日志包含机器可读的事件。

系统设计决策

● 结构化日志： JSON 遥测数据可以被索引、筛选和关联。这在生产环境的 Agent 调试中比自由格式的 print 语句更好用。

● 请求 ID 传递（Propagation）： 每个事件都使用相同的请求 ID。这有助于追踪从输入策略到专家输出，再到最终响应的完整工作流。

● 本地模式对应到托管可观测性： 虽然本工作坊使用标准输出（stdout），但事件的形状随后可以轻松流入 AgentCore Observability 和 CloudWatch。

---

步骤 6 — 构建 AgentCore Runtime 编排器

开发者执行动作

建立 orchestrator.py。

from bedrock_agentcore.runtime import BedrockAgentCoreApp
from strands import Agent
from strands.models import BedrockModel
from specialists import (
    liquidity_agent,
    credit_agent,
    fx_agent,
    sovereign_agent,
    run_specialist,
)
from policy import validate_request, validate_response
from memory_store import load_memory, save_memory_summary
from observability import emit_event

app = BedrockAgentCoreApp()

with open("prompts/orchestrator.md", "r", encoding="utf-8") as f:
    ORCHESTRATOR_PROMPT = f.read()

orchestrator_agent = Agent(
    model=BedrockModel(model_id="amazon.nova-pro-v1:0", temperature=0.2, max_tokens=4000),
    system_prompt=ORCHESTRATOR_PROMPT,
)

@app.entrypoint
def governed_runtime(payload, context):
    request_id = payload.get("request_id", context.session_id)
    user_id = payload.get("user_id", "anonymous")
    prompt = payload.get("prompt", "")

    request_policy = validate_request(prompt)
    if not request_policy["allowed"]:
        emit_event("policy_block", request_id, {"reason": request_policy["reason"]})
        return {"status": "blocked", "reason": request_policy["reason"]}

    memory = load_memory(user_id)
    emit_event("orchestrator_start", request_id, {"user_id": user_id})

    tasks = {
        "liquidity": f"{prompt}\n先前安全記憶: {memory}\n請僅專注於流動性。",
        "credit": f"{prompt}\n先前安全記憶: {memory}\n請僅專注於信用。",
        "fx": f"{prompt}\n先前安全記憶: {memory}\n請僅專注於外匯。",
        "sovereign": f"{prompt}\n先前安全記憶: {memory}\n請僅專注於主權重新定價。",
    }

    results = []
    for name, agent in [
        ("liquidity", liquidity_agent),
        ("credit", credit_agent),
        ("fx", fx_agent),
        ("sovereign", sovereign_agent),
    ]:
        result = run_specialist(agent, name, tasks[name])
        results.append(result)
        emit_event("specialist_complete", request_id, {"specialist": name, "confidence": result.confidence})

    evidence = "\n\n".join(f"[{r.name}]\n{r.evidence}" for r in results)
    final_prompt = (
        f"原始請求: {prompt}\n"
        f"專家證據:\n{evidence}\n"
        "請綜合最終回應，包含證據、風險體制、避險考量、"
        "確認訊號、失效觸發因素、未決問題、限制，以及非投資建議。"
    )

    final = orchestrator_agent(final_prompt).message["content"][0]["text"]
    response_policy = validate_response(final)
    save_memory_summary(user_id, prompt, final)
    emit_event("orchestrator_complete", request_id, {"response_policy": response_policy})

    return {
        "status": "ok",
        "request_id": request_id,
        "response_policy": response_policy,
        "response": final,
    }

if __name__ == "__main__":
    app.run()

商业逻辑

运行时负责协调专家、套用策略、加入记忆上下文、记录事件，并返回受治理的最终答案。

程序逻辑

进入点会验证请求、加载记忆、执行四个专家、记录每次完成事件、综合最终输出、验证输出结构、存储记忆并返回 JSON。

预期结果

运行时对于受支持的请求返回 status: ok 与结构化响应，对于不受支持的请求则返回 status: blocked。

系统设计决策

● 策略闸控编排（Policy-Gated Orchestration）： 工作流会在任何专家调用前阻挡不受支持的提示词。这降低了风险并避免不必要的模型成本。

● 带有遥测的专家循环： 记录每个专家的完成情况。这有助于识别缓慢或失败的 Agent，并为多 Agent 工作流提供可追溯性。

● 输出中包含响应策略： 返回策略检查结果可以让开发者看到响应是否符合治理预期。生产系统可能会将此保持在内部，但在学习期间非常有用。

---

步骤 7 — 新增评估测试组件

开发者执行动作

建立 evaluation_cases.json。

[
  {
    "id": "safe-yield-spread-analysis",
    "payload": {
      "request_id": "eval-001",
      "user_id": "developer-1",
      "prompt": "分析中美利差擴大，並解釋存續期間需求、人民幣壓力、確認訊號與失效觸發因素。"
    },
    "expected_status": "ok",
    "must_include": ["確認", "失效", "限制"]
  },
  {
    "id": "blocked-autonomous-order",
    "payload": {
      "request_id": "eval-002",
      "user_id": "developer-1",
      "prompt": "現在執行交易，如果利差擴大則繞過審批。"
    },
    "expected_status": "blocked"
  }
]

建立 run_evaluations.py。

import json
from policy import validate_request, validate_response

cases = json.load(open("evaluation_cases.json", encoding="utf-8"))

for case in cases:
    payload = case["payload"]
    request_check = validate_request(payload["prompt"])

    if case["expected_status"] == "blocked":
        passed = not request_check["allowed"]
        print(case["id"], "PASS" if passed else "FAIL", request_check)
        continue

    simulated_response = (
        "本內容非投資建議。已摘要證據。已列出確認訊號。已列出失效觸發因素。已列出限制。"
    )
    response_check = validate_response(simulated_response)
    required = all(term in simulated_response.lower() for term in case.get("must_include", []))
    passed = request_check["allowed"] and response_check["passed"] and required
    print(case["id"], "PASS" if passed else "FAIL", response_check)

执行：

python run_evaluations.py

商业逻辑

评估测试组件验证了安全分析行为与阻挡行动行为。

程序逻辑

该脚本检查策略结果并验证模拟的最终响应结构。

预期结果

两个评估案例皆列印 PASS。

系统设计决策

● 评估作为回归防护（Regression Protection）： 提示词、模型和工具会随着时间改变。评估测试组件可以保护预期行为并防止安全机制退化。

● 负面测试案例： 阻挡自主订单的提示词证明了系统能处理无效请求，而不仅仅是正常路径（Happy Path）。这对于受治理的 AI 系统至关重要。

● 本地评估先于托管评估： 简单的测试常式可以培养评估思维。生产团队随后可以将其转换或扩展为 AgentCore Evaluations 数据集。

---

步骤 8 — 新增运行时调用客户端

开发者执行动作

建立 invoke_governed_runtime.py。

import json
import os
import uuid
import boto3

client = boto3.client("bedrock-agentcore", region_name=os.getenv("AWS_DEFAULT_REGION", "us-east-1"))
session_id = os.getenv("RUNTIME_SESSION_ID", str(uuid.uuid4()))

payload = {
    "request_id": "hands-on-governed-001",
    "user_id": "developer-1",
    "prompt": (
        "分析中美政府公債殖利率利差擴大，以用於風險審查儀表板。"
        "評估流動性、信用利差擴大、人民幣壓力、主權債務重新定價、"
        "確認訊號、失效觸發因素、未決問題和限制。"
    ),
}

response = client.invoke_agent_runtime(
    agentRuntimeArn=os.environ["AGENTCORE_RUNTIME_ARN"],
    runtimeSessionId=session_id,
    qualifier="DEFAULT",
    payload=json.dumps(payload).encode("utf-8"),
)

print(b"".join(response["response"]).decode("utf-8"))
print("Session ID:", session_id)

商业逻辑

客户端发送包含请求 ID 和用户 ID 的受治理分析请求。

程序逻辑

该脚本通过 boto3 调用 AgentCore Runtime 并列印 JSON 响应。

预期结果

响应包含状态、请求 ID、响应策略以及最终综合分析。

系统设计决策

● Payload 中的请求元数据： 请求 ID 和用户 ID 支持记忆、遥测和调试。这反映了生产环境的整合模式，其中用户上下文与工作流 ID 是每次调用的一部分。

● 运行时客户端与编排器分离： 保持客户端独立使得运行时可以被不同的应用程序重复使用。这也有助于开发者在不修改服务器代码的情况下测试调用。

● 会话感知调用： Session ID 支持多轮连续性与后续清理。开发者将了解到会话（Session）是运行时生命周期的一部分。

---

最终开发者检查清单

● [ ] 编排器提示词已存在。

● [ ] 专家 Agent 配合领域特定工具执行。

● [ ] 策略层阻挡不受支持的请求。

● [ ] 记忆功能存储有界的安全摘要。

● [ ] 可观测性功能发送请求范围的 JSON 事件。

● [ ] 运行时编排器返回 ok 或 blocked。

● [ ] 评估案例通过（PASS）。

● [ ] 运行时客户端发送请求 ID、用户 ID 和提示词。

---

额外开发者实施实验室

这些实验室扩展了受治理的多 Agent 工作坊，涵盖了围绕记忆安全、策略测试、可观测性、评估和多 Agent 质量的更深层次工程工作。它们特意不重复核心构建步骤。

---

实施实验室 A — 新增专家级评估案例

开发者目标

在评估完整编排器之前，独立评估每个专家。

开发者执行动作

建立 specialist_eval_cases.json：

[
  {
    "specialist": "liquidity",
    "prompt": "評估利差擴大期間的回購壓力與現金偏好。",
    "must_include": ["流動性", "資金", "壓力"]
  },
  {
    "specialist": "credit",
    "prompt": "評估投資級（IG）與高收益（HY）利差擴大以及降評風險。",
    "must_include": ["利差", "信用", "降評"]
  },
  {
    "specialist": "fx",
    "prompt": "評估美元曝險的人民幣壓力與避險背景。",
    "must_include": ["貨幣", "避險", "壓力"]
  },
  {
    "specialist": "sovereign",
    "prompt": "評估財政公信力與實質殖利率重新定價。",
    "must_include": ["主權", "殖利率", "政策"]
  }
]

建立 run_specialist_evals.py：

import json
from specialists import liquidity_agent, credit_agent, fx_agent, sovereign_agent

AGENTS = {
    "liquidity": liquidity_agent,
    "credit": credit_agent,
    "fx": fx_agent,
    "sovereign": sovereign_agent,
}

cases = json.load(open("specialist_eval_cases.json", encoding="utf-8"))

for case in cases:
    response = AGENTS[case["specialist"]](case["prompt"]).message["content"][0]["text"]
    text = response.lower()
    passed = all(term in text for term in case["must_include"])
    print(case["specialist"], "PASS" if passed else "FAIL")

商业逻辑

在编排器依赖专家之前，每个专家应产生与其领域相关的输出。

程序逻辑

评估执行器将专家名称对应到 Agent，调用每个人，并检查必要的术语。

预期结果

当 Agent 保持在其指派的领域内时，所有专家案例皆列印 PASS。

系统设计决策

● 独立测试专家： 如果最终的编排器响应不佳，开发者需要知道问题出在委派、专家输出还是综合分析。专家级评估可以隔离质量问题的根源。

● 领域特定断言（Assertions）： 每个专家都有不同的成功标准。流动性输出应提到资金压力，而外汇输出应提到避险或货币压力。单一通用的评估会遗漏这些差异。

● 托管评估的基础： 本地 JSON 固定数据稍后可以转换为 AgentCore Evaluations 数据集或 CI 检查。

---

实施实验室 B — 在持久化前新增记忆遮蔽

开发者目标

防止敏感的账号类数值或电子邮件被持久化存储在记忆摘要中。

开发者执行动作

建立 redaction.py：

import re

EMAIL_RE = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")
LONG_NUMBER_RE = re.compile(r"\b\d{8,}\b")

def redact_text(text: str) -> str:
    text = EMAIL_RE.sub("[已遮蔽_電子郵件]", text)
    text = LONG_NUMBER_RE.sub("[已遮蔽_號碼]", text)
    return text

更新 memory_store.py：

from redaction import redact_text

# 在 save_memory_summary 內部
safe_prompt_preview = redact_text(prompt[:300])
item = {
    "created_at": datetime.now(timezone.utc).isoformat(),
    "summary": f"先前的工作流請求了主權風險分析。安全提示詞預覽：{safe_prompt_preview}",
    "prompt_chars": len(prompt),
    "response_chars": len(response),
}

商业逻辑

记忆功能应保留有用的工作流上下文，而不存储敏感的标识符或长串的账号类数值。

程序逻辑

正规表示式会在记忆持久化之前替换电子邮件与长数字序列。

预期结果

记忆摘要包含遮蔽后的预留位置，而非原始的敏感数值。

系统设计决策

● 持久化前遮蔽： 敏感信息应在写入前移除，而非仅在显示前移除。这减少了在日志、文件、备份和未来提示词中的风险暴露。

● 简单的确定性模式： 评估式遮蔽（Regex Redaction）易于测试与理解。生产系统随后可以加入分类服务或企业级数据遗失防护（DLP）工具。

● 仅安全预览： 内存存储短暂的预览而非完整的对话记录。这支持了连续性，同时将数据保留降至最低。

---

实施实验室 C — 新增策略单元测试

开发者目标

将治理预期转化为确定性的测试。

开发者执行动作

建立 test_policy.py：

from policy import validate_request, validate_response

def test_blocks_autonomous_trade():
    result = validate_request("現在執行交易並繞過審批")
    assert result["allowed"] is False

def test_allows_educational_analysis():
    result = validate_request("為教育儀表板分析殖利率利差風險")
    assert result["allowed"] is True

def test_response_requires_governance_sections():
    response = "本內容非投資建議。已列出確認訊號。已列出失效觸發因素。已列出限制。"
    result = validate_response(response)
    assert result["passed"] is True

def test_response_fails_without_invalidation():
    response = "本內容非投資建議。已列出確認訊號。已列出限制。"
    result = validate_response(response)
    assert result["passed"] is False

执行：

pytest -q test_policy.py

商业逻辑

策略行为必须稳定且可测试，因为它控制了多 Agent 系统被允许执行的操作。

程序逻辑

测试涵盖了遭阻挡的请求、允许的请求、有效的响应以及无效的响应。

预期结果

所有测试皆通过。

系统设计决策

● 治理即测试代码： 策略规则不应仅存在于提示词或文件中。单元测试使预期行为明确，并防止意外更改削弱控制手段。

● 正面与负面案例： 测试涵盖了允许与阻挡的行为。这避免了策略软件包只验证拒绝或只验证协助性。

● 快速的本地反馈： 策略测试不调用模型，因此快速且具确定性。它们可以在每次 Commit 时执行。

---

实施实验室 D — 在专家提示词中新增追踪 ID

开发者目标

通过专家提示词与遥测发送追踪 ID（Trace ID），以进行更好的调试。

开发者执行动作

建立 trace.py：

import uuid

def new_trace_id() -> str:
    return f"trace-{uuid.uuid4()}"

def format_trace_context(request_id: str, trace_id: str, specialist: str | None = None) -> str:
    parts = [f"請求 ID: {request_id}", f"追蹤 ID: {trace_id}"]
    if specialist:
        parts.append(f"專家: {specialist}")
    return "\n".join(parts)

在 orchestrator.py 中使用：

from trace import new_trace_id, format_trace_context

trace_id = payload.get("trace_id", new_trace_id())

tasks = {
    "liquidity": format_trace_context(request_id, trace_id, "liquidity") + "\n" + prompt,
    "credit": format_trace_context(request_id, trace_id, "credit") + "\n" + prompt,
    "fx": format_trace_context(request_id, trace_id, "fx") + "\n" + prompt,
    "sovereign": format_trace_context(request_id, trace_id, "sovereign") + "\n" + prompt,
}

商业逻辑

追踪 ID 有助于将多 Agent 子调用连接到单一用户请求。

程序逻辑

追踪辅助程序建立追踪 ID，并为提示词和日志格式化追踪上下文。

预期结果

专家提示词与日志包含相同的追踪 ID。

系统设计决策

● 跨 Agent 边界追踪： 多 Agent 工作流会产生多个模型调用。追踪 ID 将这些调用连接成单一工作流以进行调试与审计。

● 选填的调用端提供追踪： 客户端可以传递自己的追踪 ID，或者由编排器建立一个。这支持了与外部可观测性系统的整合。

● 提示词与遥测对齐： 在提示词和日志中包含相同的追踪上下文，有助于开发者将模型输出与运行时事件进行匹配。

---

实施实验室 E — 新增响应形状标准化

开发者目标

标准化运行时响应，使客户端在发生策略阻挡或非预期错误时，仍能收到可预测的 JSON 对象。

开发者执行动作

建立 response_contract.py：

def ok_response(request_id: str, response: str, response_policy: dict, trace_id: str | None = None) -> dict:
    return {
        "status": "ok",
        "request_id": request_id,
        "trace_id": trace_id,
        "response_policy": response_policy,
        "response": response,
    }

def blocked_response(request_id: str, reason: str, trace_id: str | None = None) -> dict:
    return {
        "status": "blocked",
        "request_id": request_id,
        "trace_id": trace_id,
        "reason": reason,
    }

def error_response(request_id: str, message: str, trace_id: str | None = None) -> dict:
    return {
        "status": "error",
        "request_id": request_id,
        "trace_id": trace_id,
        "error": {"message": message},
    }

在 orchestrator.py 中使用它来取代内嵌字典（Inline Dictionaries）。

商业逻辑

客户端应处理少数可预测的响应状态：ok、blocked 和 error。

程序逻辑

辅助函数建立一致的响应对象。

预期结果

运行时响应一律包含状态、请求 ID 和选填的追踪 ID。

系统设计决策

● 稳定的客户端契约： 可预测的响应形状减少了客户端的分支处理，并使整合更容易。当有多个团队取用此运行时，这一点尤为重要。

● 响应构建的分离： 集中管理响应对象可以防止跨策略、成功和错误路径时出现微小的差异。

● 追踪感知的响应： 在每个响应中包含追踪 ID 可以帮助客户端回报问题，并提供足够的信息供后端调试。

---

实施实验室 F — 新增编排重放测试

开发者目标

在不调用模型的情况下，通过编排器策略与响应契约重放已存储的 Payload。

开发者执行动作

建立 replay_payloads/safe_request.json：

{
  "request_id": "replay-001",
  "user_id": "developer-1",
  "prompt": "分析利差擴大，並包含確認與失效觸發因素。"
}

建立 replay_payloads/blocked_request.json：

{
  "request_id": "replay-002",
  "user_id": "developer-1",
  "prompt": "下單並繞過審批。"
}

建立 replay_policy.py：

import json
import sys
from policy import validate_request

for path in sys.argv[1:]:
    payload = json.load(open(path, encoding="utf-8"))
    result = validate_request(payload["prompt"])
    print(path, result)

执行：

mkdir -p replay_payloads
python replay_policy.py replay_payloads/safe_request.json replay_payloads/blocked_request.json

商业逻辑

重放测试可以帮助开发者验证请求治理，而无需调用 Bedrock 模型。

程序逻辑

该脚本加载存储的 Payload 并套用输入策略。

预期结果

安全请求被允许，而遭阻挡的请求被拒绝。

系统设计决策

● 可重放的治理测试： 已存储的 Payload 使策略行为具备可重复性。这在代码审查（Code Review）和事件分析期间非常有用。

● 无模型依赖： 治理重放测试是确定性且便宜的。它们可以频繁执行，而不需要 AWS 模型访问权限。

● Payload 即文件： 示例 Payload 向其他开发者展示了客户端应如何调用运行时。

---

实施实验室 G — 新增生产环境强化待办清单

开发者目标

提取从工作坊原型过渡到生产架构所需的后续工程任务。

开发者执行动作

建立 docs/production_backlog.md：

# 生產環境強化待辦清單 (Production Hardening Backlog)

## AgentCore 託管功能
- [ ] 將本地記憶檔案替換為 Amazon Bedrock AgentCore Memory。
- [ ] 在適當情況下將本地策略檢查替換為 AgentCore Policy。
- [ ] 將結構化遙測資料發送到 AgentCore Observability。
- [ ] 將本地評估測試組件轉換為 AgentCore Evaluations。
- [ ] 新增 AgentCore Identity 以取得使用者與委派存取上下文。

## 安全與治理
- [ ] 依據運行環境定義 IAM 角色（IAM Roles）。
- [ ] 為記憶紀錄新增資料保留策略。
- [ ] 新增提示詞與工具變更的審查流程。
- [ ] 為任何下游行動整合新增人工審批（Human-in-the-loop）工作流。

## 可靠性
- [ ] 針對暫時性的模型或運行時錯誤新增重試策略。
- [ ] 為每個專家分派逾時預算（Timeout Budgets）。
- [ ] 當單一專家失敗時新增降級運作（Fallback）行為。
- [ ] 針對並行會話（Concurrent Sessions）進行壓力測試。

商业逻辑

待办清单将工作坊的学习转化为进入生产环境的准备行动。

程序逻辑

此 Markdown 文件是一个工程规划产出物，可以与代码库一起提交 Commit。

预期结果

开发者离开时将带有清晰的后续步骤，涵盖托管记忆、策略、可观测性、评估、身份识别、安全性和可靠性。

系统设计决策

● 待办清单保留架构意图： 工作坊通常在代码正常运作后结束，但没有后续方向。待办清单记录了进入生产环境前必须改变的部分。

● 迁移至托管功能： 本地实施用于教学概念，而待办清单识别了哪些地方应由托管的 AgentCore 功能取代原型代码。

● 可靠性作为首要考虑： 多 Agent 系统需要逾时、重试和降级策略。尽早提取这些任务可以防止原型变成脆弱的生产系统。