AWS Builder Workshop

使用 AgentCore 與 Strands 建構：執行期主權風險代理人開發者工作坊

使用 AgentCore 與 Strands 建構：執行期主權風險代理人開發者工作坊

目標受眾： 專業 Python/AWS 開發者

時長： 2 小時

主要 AWS AI 服務： Amazon Bedrock AgentCore Runtime、Strands Agents、Amazon Bedrock 模型

工作坊風格： 開發者實戰動手建構，無需簡報

專案輸出： 一個具備本地驗證、部署指令碼、boto3 呼叫用戶端、串流變體、工作階段清理以及大容量 Payload 處理器的全功能執行期代理人（Runtime Agent）。

本工作坊僅供教育工程教學使用。範例領域為主權風險分析，但輸出內容不構成任何財務、投資、法律、稅務或交易建議。

工作坊摘要

開發者將使用 Strands 與 AgentCore Runtime 建構一個生產級的主權風險代理人。工作坊內容涵蓋決定性（Deterministic）計算工具、Payload 驗證、冒煙測試、部署、boto3 用戶端呼叫、串流回應、大容量 base64 Payload 處理以及工作階段清理。參與者在結束時將獲得可重複使用的執行期模式，用於建構安全、具備可觀測性且感知工作階段（Session-aware）的 AI 服務，並在 AWS 上將模型推理與可測試的 Python 邏輯相結合，以滿足現代企業開發者的工作流程。

1. 開發者學習目標

在本工作坊結束時，開發者將能夠：

  1. 建構一個由 Amazon Bedrock 模型支援的 Strands 代理人。
  2. 為以 LLM 驅動的代理人添加決定性的 Python 工具。
  3. 將代理人封裝至 Amazon Bedrock AgentCore Runtime 進入點（Entrypoint）。
  4. 使用 AgentCore 入門工具包（Starter Toolkit）封裝並啟動執行期。
  5. 使用穩定的工作階段 ID（Session ID）透過 boto3 呼叫執行期。
  6. 為需要即時回應的用戶端實作串流執行期進入點。
  7. 建構一個可處理 base64 編碼的 Excel 和圖片輸入的大容量 Payload 進入點。
  8. 在部署前加入本地 Payload 合約驗證與冒煙測試。

2. 本工作坊建構的架構

開發者筆記型電腦
  ├─ app.py                         # 同步 AgentCore Runtime 進入點
  ├─ app_streaming.py               # 串流 Runtime 進入點
  ├─ app_large_payload.py           # 多模態 / 大容量 Payload Runtime 進入點
  ├─ deploy_runtime.py              # 封裝並啟動執行期
  ├─ invoke_runtime.py              # boto3 用戶端呼叫
  ├─ stop_session.py                # 顯式工作階段清理
  ├─ validate_payload.py            # 本地 Payload 合約驗證
  ├─ smoke_test.py                  # 本地冒煙測試
  ├─ prompts/system.md              # 受控的系統提示詞
  └─ requirements.txt
AWS
  ├─ 透過 Strands 存取的 Amazon Bedrock 模型
  ├─ Amazon Bedrock AgentCore Runtime
  ├─ 由工具包建立的 Amazon ECR 映像檔
  ├─ IAM 執行角色
  └─ 雲端日誌 / 執行期輸出

3. 2 小時實戰議程

時間（分鐘）模組實戰輸出
0–10環境檢查確認 AWS 身分、區域與 Python 環境
10–25專案基礎架構建立檔案、依賴項目與提示詞合約
25–45Strands 代理人代理人使用 Bedrock 模型與決定性工具
45–60本地驗證在本地執行 Payload 驗證器與冒煙測試
60–80AgentCore Runtime 部署產生執行期 ARN
80–95執行期呼叫boto3 用戶端呼叫感知工作階段的代理人
95–110串流變體實作非同步串流進入點
110–120大容量 Payload 與清理新增 base64 檔案 Payload 與停止工作階段指令碼

4. 先決條件

python --version     # 建議版本：3.11+
aws --version
aws sts get-caller-identity
export AWS_DEFAULT_REGION=us-east-1

建立並啟用虛擬環境：

python -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip

步驟 1 — 建立專案基礎架構

開發者操作

mkdir -p agentcore-strands-runtime/prompts
cd agentcore-strands-runtime
cat > requirements.txt <<'EOF'
bedrock-agentcore
bedrock-agentcore-starter-toolkit
strands-agents
strands-agents-tools
boto3
botocore
pytest
EOF

建立系統提示詞：

cat > prompts/system.md <<'EOF'
您是一位專業的主權風險工程助理。
請分析殖利率價差、信用利差擴大、主權債務重新定價、外匯壓力、
資金流動性、存續期間需求、資本流動以及防禦性輪動。
請務必包含以下章節：
1. 觀察 (Observation)
2. 推理 (Reasoning)
3. 風險隱含意義 (Risk implication)
4. 避險考量 (Hedge consideration)
5. 確認訊號 (Confirmation signals)
6. 失效觸發條件 (Invalidation triggers)
7. 限制 (Limitations)
請使用工具進行數值計算。請勿提供投資建議或自主交易指令。
EOF

安裝依賴項目：

pip install -r requirements.txt

系統設計決策

提示詞檔案作為版本控制的行為： 系統提示詞是執行期合約的一部分。將其保存在 prompts/system.md 中，可讓開發者像審查程式碼變更一樣審查行為變更。這非常重要，因為代理人的輸出風格、安全邊界和必要章節會直接影響下游用戶端、測試與評估。 明確的依賴檔案： 執行期部署必須具備可重複性。requirements.txt 定義了執行期所需的精確 Python 套件介面。這也簡化了部署封裝、容器建構以及工作階段的疑難排解，因為每位參與者安裝的依賴項目完全相同。 * 雲端部署前的微型架構： 開發者先在本地建立檔案並驗證行為，然後才部署到雲端。這減少了雲端除錯的雜音。如果缺少提示詞檔案或 Python 匯入失敗，錯誤會在執行期被封裝和啟動之前在本地被擷取到。

預期結果

專案中包含一個依賴檔案和一個受控的系統提示詞，該提示詞將由 Strands 代理人讀取。

步驟 2 — 建構同步 Strands 代理人執行期

開發者操作

建立 app.py。

from strands import Agent, tool
from strands.models import BedrockModel
from strands_tools import calculator
from bedrock_agentcore.runtime import BedrockAgentCoreApp
app = BedrockAgentCoreApp()
with open("prompts/system.md", "r", encoding="utf-8") as f:
    SYSTEM_PROMPT = f.read()
@tool
def yield_spread_bps(us_yield: float, peer_yield: float) -> str:
    """計算兩個殖利率之間的基點（bps）價差。"""
    spread = (us_yield - peer_yield) * 100
    return f"殖利率價差為 {spread:.1f} 基點。"
@tool
def fx_hedge_notional(exposure_usd: float, hedge_ratio: float) -> str:
    """根據 USD 風險敞口和避險比率計算避險名目本金。"""
    hedge = exposure_usd * hedge_ratio
    return f"外匯避險名目本金為 USD {hedge:,.2f}。"
@tool
def stress_loss_notional(position_notional: float, spread_move_bps: float, duration: float) -> str:
    """估算由價差變動引起的存續期間驅動價格影響。"""
    loss = position_notional * duration * (spread_move_bps / 10000)
    return f"預估存續期間影響為 USD {loss:,.2f}。"
agent = Agent(
    model=BedrockModel(
        model_id="amazon.nova-pro-v1:0",
        temperature=0.2,
        max_tokens=4000,
    ),
    tools=[calculator, yield_spread_bps, fx_hedge_notional, stress_loss_notional],
    system_prompt=SYSTEM_PROMPT,
)
@app.entrypoint
def sovereign_runtime(payload, context):
    prompt = payload.get("prompt", "")
    if not prompt.strip():
        return {"error": "缺少必要欄位：prompt"}
    request_id = payload.get("request_id", context.session_id)
    request = (
        f"請求 ID: {request_id}\n"
        f"執行期工作階段: {context.session_id}\n"
        f"使用者請求: {prompt}"
    )
    response = agent(request)
    return response.message["content"][0]["text"]
if __name__ == "__main__":
    app.run()

商業邏輯

代理人分析主權風險提示詞，並使用工具計算基點價差、外匯避險名目本金以及存續期間壓力影響。模型處理定性綜合分析，而決定性計算則留在 Python 中執行。

程式碼邏輯

BedrockAgentCoreApp() 建立執行期應用程式包裝器。 BedrockModel() 設定 Strands 所使用的 Amazon Bedrock 模型。 @tool 將 Python 函式公開給 Strands 代理人。 @app.entrypoint 標記可呼叫的執行期處理器（Handler）。 * 處理器驗證 prompt，加入請求與工作階段上下文，呼叫代理人，並回傳最終文字。

預期結果

執行 python app.py 會啟動一個本地與 AgentCore 相容的執行期處理程序。部署後，同一個進入點將接收來自 AgentCore Runtime 的 JSON Payload。

系統設計決策

用於定量邏輯的決定性工具： LLM 非常適合用於綜合分析，但財務計算應該是決定性且可測試的。透過將殖利率價差、避險名目本金和存續期間壓力數學運算移至 Python 工具中，開發者可以獲得可重複的輸出，並能獨立於模型行為之外對計算進行單元測試。 請求 ID 與工作階段上下文： 進入點將請求與工作階段 ID 注入模型提示詞中。這為回應提供了運作上下文，並有助於工程師關聯執行期日誌、用戶端呼叫和使用者工作流程。這也為生產環境的追蹤（Tracing）建立了一個乾淨的模式。 * 低溫度的 Bedrock 模型： 代理人使用 temperature=0.2 來減少回應的變異性。專業的開發者工作坊應該教授穩定的工程模式，而非創意的提示詞實驗。較低的變異性有助於改善冒煙測試、展示以及未來的評估基準。

步驟 3 — 加入本地 Payload 驗證

開發者操作

建立 validate_payload.py。

REQUIRED_FIELDS = ["prompt"]
OPTIONAL_FIELDS = ["request_id", "user_id", "excel_data", "image_data"]
def validate_payload(payload: dict) -> tuple[bool, list[str]]:
    errors = []
    for field in REQUIRED_FIELDS:
        if field not in payload or not str(payload[field]).strip():
            errors.append(f"缺少必要欄位：{field}")
    unknown = set(payload.keys()) - set(REQUIRED_FIELDS) - set(OPTIONAL_FIELDS)
    for field in sorted(unknown):
        errors.append(f"未知欄位：{field}")
    return len(errors) == 0, errors
if __name__ == "__main__":
    sample = {"prompt": "分析美中殖利率價差", "request_id": "demo-001"}
    ok, errors = validate_payload(sample)
    print("有效性", ok)
    print("錯誤", errors)

商業邏輯

驗證器在執行期呼叫之前強制執行請求合約。這可以防止用戶端傳送格式錯誤的請求，並使除錯更容易。

程式碼邏輯

該函式檢查必要欄位、拒絕未知欄位，並回傳一個布林值與錯誤列表。它可以被測試、用戶端或執行期程式碼匯入。

預期結果

python validate_payload.py
# 有效性 True
# 錯誤 []

系統設計決策

部署前的 Payload 合約： 開發者應在呼叫雲端執行期之前在本地驗證 API 合約。這減少了因缺少提示詞或欄位拼錯而導致的執行期呼叫失敗。這也為以後建立更強大的綱要（Schema）奠定了基礎。 明確的未知欄位偵測： 拒絕未知欄位有助於及早發現整合錯誤。如果沒有這項機制，用戶端可能會誤以為元資料（Metadata）已被使用，而執行期其實默默忽略了它。清晰的驗證能改善開發者回饋。 * 可重複使用的驗證模組： 驗證器是一個模組，而不僅僅是一個指令碼。它可以在單元測試、CLI 用戶端和執行期進入點中重複使用。這避免了在多個檔案中重複編寫合約邏輯。

步驟 4 — 加入冒煙測試

開發者操作

建立 smoke_test.py。

from validate_payload import validate_payload
from app import yield_spread_bps, fx_hedge_notional, stress_loss_notional
def test_payload_validation():
    ok, errors = validate_payload({"prompt": "分析價差"})
    assert ok
    assert errors == []
def test_missing_prompt_fails():
    ok, errors = validate_payload({"request_id": "x"})
    assert not ok
    assert "缺少必要欄位：prompt" in errors
def test_tool_outputs():
    assert "210.0" in yield_spread_bps(4.25, 2.15)
    assert "15,000,000.00" in fx_hedge_notional(25_000_000, 0.6)
    assert "125,000.00" in stress_loss_notional(10_000_000, 25, 5)

執行：

pytest -q

商業邏輯

冒煙測試在部署代理人之前，驗證請求合約與決定性計算工具。

程式碼邏輯

測試直接匯入驗證與工具函式。它們不會呼叫模型，從而保持測試的高速與決定性。

預期結果

3 passed

系統設計決策

優先測試決定性部分： 模型的輸出可能會有所不同，但工具數學運算和 Payload 驗證不應該改變。測試決定性元件可以為開發者提供快速回饋，並在雲端部署前隔離錯誤。 冒煙測試中無模型依賴： 冒煙測試不會呼叫 Amazon Bedrock。這避免了成本、憑證、模型延遲和非決定性斷言（Assertions）的問題。其目標是驗證本地軟體合約。 * 工作坊友善的失敗訊息： 簡單的斷言使失敗原因一目了然。在兩小時的工作坊中，除錯必須快速且集中。這些測試能及早發現常見的設定與邏輯錯誤。

步驟 5 — 部署 AgentCore Runtime

開發者操作

建立 deploy_runtime.py。

from boto3.session import Session
from bedrock_agentcore_starter_toolkit import Runtime
region = Session().region_name or "us-east-1"
runtime = Runtime()
runtime.configure(
    entrypoint="app.py",
    auto_create_execution_role=True,
    auto_create_ecr=True,
    requirements_file="requirements.txt",
    region=region,
    agent_name="sovereign-runtime-hands-on",
)
result = runtime.launch()
print("AGENTCORE_RUNTIME_ARN=", result.agent_arn)
print("AGENTCORE_RUNTIME_ID=", result.agent_id)

執行：

python deploy_runtime.py
export AGENTCORE_RUNTIME_ARN="貼上代理人執行期 ARN"

商業邏輯

本地代理人成為一個受管的執行期端點，應用程式可以呼叫該端點。

程式碼邏輯

Runtime().configure() 封裝進入點與依賴項目。launch() 部署執行期並回傳識別碼。

預期結果

指令碼印出 AgentCore Runtime ARN 與執行期 ID。

系統設計決策

利用工具包提高實戰速度： 入門工具包處理了部署機制，使工作坊能夠專注於執行期設計，而非容器內部細節。開發者仍然可以看到產生的執行期 ARN，並可以在事後檢查生成的 AWS 資源。 先部署單一執行期進入點： 僅部署 app.py 可以保持首次雲端部署的簡單性。一旦同步呼叫正常運作，開發者就可以在減少未知變數的情況下部署串流或大容量 Payload 變體。 * 使用環境變數儲存 ARN： 將執行期 ARN 匯出為環境變數，可以將部署與呼叫指令碼解耦。這模擬了生產環境模式，其中部署輸出會提供給用戶端設定或 CI/CD 變數。

步驟 6 — 使用 boto3 呼叫執行期

開發者操作

建立 invoke_runtime.py。

import json
import os
import uuid
import boto3
from validate_payload import validate_payload
region = os.getenv("AWS_DEFAULT_REGION", "us-east-1")
agent_arn = os.environ["AGENTCORE_RUNTIME_ARN"]
session_id = os.getenv("RUNTIME_SESSION_ID", str(uuid.uuid4()))
payload = {
    "request_id": "hands-on-001",
    "prompt": (
        "分析美國 10 年期 4.25 對比中國 10 年期 2.15。"
        "USD 風險敞口為 25000000 且避險比率為 0.60。"
        "部位名目本金為 10000000，價差變動為 25 bps，存續期間為 5。"
        "請包含確認訊號與失效觸發條件。"
    ),
}
ok, errors = validate_payload(payload)
if not ok:
    raise ValueError(errors)
client = boto3.client("bedrock-agentcore", region_name=region)
res = client.invoke_agent_runtime(
    agentRuntimeArn=agent_arn,
    runtimeSessionId=session_id,
    qualifier="DEFAULT",
    payload=json.dumps(payload).encode("utf-8"),
)
body = b"".join(res["response"]).decode("utf-8")
print(body)
print("工作階段 ID:", session_id)

執行：

python invoke_runtime.py

商業邏輯

用戶端請求結構化的主權風險分析，並提供應觸發計算工具的數值。

程式碼邏輯

指令碼驗證 Payload、建立 boto3 AgentCore 用戶端、傳送 JSON 位元組、組合執行期回應區塊（Chunks），並印出工作階段 ID。

預期結果

回應包含分析內容以及計算出的數值，例如基點價差、避險名目本金或存續期間影響。

系統設計決策

用戶端驗證： 用戶端在傳送請求前進行驗證。這可以及早發現錯誤，並避免不必要的執行期呼叫。生產環境的用戶端可以共用相同的驗證邏輯或使用正式的 JSON Schema。 工作階段 ID 重複使用： 指令碼會印出工作階段 ID，以便開發者可以將其重複用於多輪（Multi-turn）測試。這展示了 AgentCore 工作階段如何支援跨呼叫的上下文工作流程。 * 回應正規化： 指令碼將回應區塊組合成單一字串。這種抽象化保持了下游用戶端的簡單性，同時仍能與事件風格的執行期回應相容。

步驟 7 — 加入串流執行期變體

開發者操作

建立 app_streaming.py。

from strands import Agent, tool
from strands.models import BedrockModel
from bedrock_agentcore.runtime import BedrockAgentCoreApp
app = BedrockAgentCoreApp()
@tool
def credit_spread_widening() -> str:
    """回傳信用利差擴大的核心驅動因素。"""
    return "驅動因素包括資金壓力、流動性撤出、調降評級風險以及防禦性部位輪動。"
agent = Agent(
    model=BedrockModel(model_id="amazon.nova-pro-v1:0", temperature=0.2, max_tokens=4000),
    tools=[credit_spread_widening],
    system_prompt="串流輸出簡明的主權風險分析，包含證據、確認與失效章節。",
)
@app.entrypoint
async def stream_runtime(payload, context):
    prompt = payload.get("prompt", "")
    if not prompt:
        yield {"type": "error", "message": "缺少 prompt"}
        return
    request = f"工作階段: {context.session_id}\n{prompt}"
    async for event in agent.stream_async(request):
        if "data" in event:
            yield event["data"]
if __name__ == "__main__":
    app.run()

商業邏輯

串流功能允許用戶端在分析內容生成時同步顯示，這對分析師儀表板和對話介面非常有用。

程式碼邏輯

進入點為 async（非同步），呼叫 agent.stream_async()，並產出（Yield）資料區塊。

預期結果

支援串流的執行期可以回傳部分區塊，而不需要等待完整回應生成完畢。

系統設計決策

用於部分輸出的非同步進入點： 串流需要一個能產出區塊的非同步進入點。這改善了使用者介面的感知延遲（Perceived latency），並向開發者展示了不同的執行期回應模式。 獨立的串流檔案： 將串流程式碼與同步執行期分開，可以輕鬆進行對比。開發者可以清楚地看到改變了什麼：改變的是進入點風格與回應處理方式，而非整個架構。 * 工具支援依然可用： 串流代理人仍然保有工具。這證明了串流是一種傳輸選擇，並不會削弱代理人的能力。

步驟 8 — 加入大容量 Payload 處理器

開發者操作

建立 app_large_payload.py。

import base64
from strands import Agent
from strands.models import BedrockModel
from bedrock_agentcore.runtime import BedrockAgentCoreApp
app = BedrockAgentCoreApp()
agent = Agent(
    model=BedrockModel(model_id="amazon.nova-pro-v1:0", temperature=0.2, max_tokens=8000),
    system_prompt=(
        "分析提供的財務文件與圖表以評估主權風險、殖利率價差、信用利差、"
        "外匯壓力、流動性、資本流動、避險考量、確認訊號與失效觸發條件。"
        "請勿提供投資建議。"
    ),
)
@app.entrypoint
def large_payload_runtime(payload, context):
    content = [{"text": f"工作階段: {context.session_id}\n{payload.get('prompt', '分析提供的資料')}"}]
    if payload.get("excel_data"):
        content.insert(0, {
            "document": {
                "format": "xlsx",
                "name": "sovereign_dataset",
                "source": {"bytes": base64.b64decode(payload["excel_data"])},
            }
        })
    if payload.get("image_data"):
        content.insert(0, {
            "image": {
                "format": "png",
                "source": {"bytes": base64.b64decode(payload["image_data"])},
            }
        })
    response = agent(content)
    return response.message["content"][0]["text"]

商業邏輯

執行期將使用者提示詞、試算表資料和圖表影像訊號整合到單一分析流程中。

程式碼邏輯

處理器對 base64 Payload 欄位進行解碼，並建構具備型別的 Bedrock 內容區塊：document、image 和 text。

預期結果

用戶端可以傳送選填的 excel_data 和 image_data 欄位，並接收整合後的分析結果。

系統設計決策

用於 JSON 傳輸的 Base64： 二進位檔案經過編碼，以便透過 JSON Payload 傳輸。這保持了用戶端合約的簡單性，並避免了工作坊期間多部分上傳（Multipart upload）的複雜性。 具備型別的內容區塊： 傳入 document 和 image 區塊可以保留模態性（Modality）。這比將所有內容轉換為文字更具維護性，並能讓模型利用文件/影像特定的專屬能力。 * 選填的 Payload 欄位： 處理器可與純文字、純文件、純影像或組合請求協同運作。這使得執行期對於多種用戶端類型更具彈性。

步驟 9 — 加入顯式工作階段清理

# stop_session.py
import os
import boto3
client = boto3.client("bedrock-agentcore", region_name=os.getenv("AWS_DEFAULT_REGION", "us-east-1"))
client.stop_runtime_session(
    agentRuntimeArn=os.environ["AGENTCORE_RUNTIME_ARN"],
    runtimeSessionId=os.environ["RUNTIME_SESSION_ID"],
    qualifier="DEFAULT",
)
print("已停止工作階段", os.environ["RUNTIME_SESSION_ID"])

系統設計決策

工作階段是受管資源： 代理人工作階段可以保留上下文並消耗資源。開發者應該在工作流程結束時顯式停止工作階段，而不是僅依賴閒置過期。 清理指令碼化： 一個小指令碼很容易在展示、測試和 CI 工作結束後執行。這鼓勵了規範的執行期維護習慣。 * 運作控制把手： 工作階段 ID 成為除錯、追蹤與清理的運作控制把手（Operational handle）。

開發者最終檢查清單

[ ] pytest -q 在本地通過。 [ ] python deploy_runtime.py 回傳執行期 ARN。 [ ] python invoke_runtime.py 回傳結構化分析。 [ ] 串流進入點編譯成功且可單獨部署。 [ ] 大容量 Payload 進入點接受選填的 excel_data 和 image_data。 [ ] 工作階段清理指令碼已通過測試。

額外開發者實戰實驗室

以下實驗室擴展了執行期工作坊，提供了更深入的開發者實作練習。它們被設計為核心兩小時建構後的選修模組，或適合希望加強部署、測試和營運規範的專業團隊的後續練習。

實戰實驗室 A — 為執行期 Payload 新增 JSON Schema 驗證

開發者目標

將簡單的欄位驗證器替換為可重複使用的 JSON Schema 驗證器，該驗證器可由用戶端、測試和執行期處理器共用。

開發者操作

安裝 jsonschema：

pip install jsonschema

建立 payload_schema.py：

RUNTIME_PAYLOAD_SCHEMA = {
    "type": "object",
    "properties": {
        "request_id": {"type": "string", "minLength": 1},
        "user_id": {"type": "string", "minLength": 1},
        "prompt": {"type": "string", "minLength": 1},
        "excel_data": {"type": "string"},
        "image_data": {"type": "string"},
        "metadata": {
            "type": "object",
            "additionalProperties": {"type": ["string", "number", "boolean", "null"]},
        },
    },
    "required": ["prompt"],
    "additionalProperties": False,
}

建立 validate_schema.py：

from jsonschema import Draft202012Validator
from payload_schema import RUNTIME_PAYLOAD_SCHEMA
validator = Draft202012Validator(RUNTIME_PAYLOAD_SCHEMA)
def validate_payload_schema(payload: dict) -> tuple[bool, list[str]]:
    errors = sorted(validator.iter_errors(payload), key=lambda e: e.path)
    messages = [f"{list(error.path)}: {error.message}" for error in errors]
    return len(messages) == 0, messages
if __name__ == "__main__":
    sample = {"prompt": "分析價差", "metadata": {"source": "lab"}}
    ok, messages = validate_payload_schema(sample)
    print(ok)
    print(messages)

商業邏輯

Schema 正式確立了執行期 API 合約。這有助於前端開發者、後端服務和測試套件在執行期支援的確切欄位上達成共識。

程式碼邏輯

Draft202012Validator 檢查 Payload 的形狀、必要欄位、資料型別與未知欄位。輔助程式回傳一個布林值與人類可讀的錯誤訊息。

預期結果

python validate_schema.py
# True
# []

系統設計決策

Schema 作為共享 API 合約： JSON Schema 比手寫驗證更精確，因為它在一個可重複使用的構件中定義了型別、必要欄位與未知欄位的行為。這允許用戶端、測試套件和執行期處理器在呼叫 AgentCore Runtime 之前驗證相同的合約。 在模型呼叫前快速失敗： 執行期呼叫可能會涉及模型延遲與成本。在呼叫代理人之前驗證 Payload 可以防止本可避免的失敗，並為開發者提供即時回饋。它還減少了由格式錯誤的用戶端 Payload 引起的異常執行期日誌。 * 可擴展的元資料物件： Schema 允許使用 metadata 物件來提供安全的營運上下文，同時封鎖任意的最上層欄位。這使得團隊在保持彈性的同時，不會讓執行期合約變得不受控制。

實戰實驗室 B — 建構可重複使用的執行期用戶端類別

開發者目標

將 boto3 呼叫、工作階段 ID、Payload 驗證和回應正規化封裝到一個可重複使用的用戶端類別中。

開發者操作

建立 runtime_client.py：

import json
import os
import uuid
import boto3
from validate_schema import validate_payload_schema
class AgentCoreRuntimeClient:
    def __init__(self, runtime_arn: str, region: str | None = None):
        self.runtime_arn = runtime_arn
        self.region = region or os.getenv("AWS_DEFAULT_REGION", "us-east-1")
        self.client = boto3.client("bedrock-agentcore", region_name=self.region)
    def invoke(self, prompt: str, session_id: str | None = None, **kwargs) -> dict:
        payload = {"prompt": prompt, **kwargs}
        ok, errors = validate_payload_schema(payload)
        if not ok:
            raise ValueError({"payload_errors": errors})
        runtime_session_id = session_id or str(uuid.uuid4())
        response = self.client.invoke_agent_runtime(
            agentRuntimeArn=self.runtime_arn,
            runtimeSessionId=runtime_session_id,
            qualifier="DEFAULT",
            payload=json.dumps(payload).encode("utf-8"),
        )
        body = b"".join(response["response"]).decode("utf-8")
        return {"session_id": runtime_session_id, "body": body}

建立 use_runtime_client.py：

import os
from runtime_client import AgentCoreRuntimeClient
client = AgentCoreRuntimeClient(os.environ["AGENTCORE_RUNTIME_ARN"])
result = client.invoke(
    prompt="分析美中殖利率價差變動，並包含確認與失效訊號。",
    request_id="client-lab-001",
    metadata={"application": "developer-lab"},
)
print(result["session_id"])
print(result["body"])

商業邏輯

應用程式團隊需要一個乾淨的執行期整合層，而不是在每個服務中重複編寫 boto3 程式碼。

程式碼邏輯

該類別驗證 Payload、建立或重複使用工作階段 ID、呼叫 AgentCore Runtime、組合回應事件，並回傳正規化的輸出。

預期結果

開發者只需兩行應用程式程式碼即可呼叫執行期。

系統設計決策

用戶端抽象化： 可重複使用的類別可以防止在各個應用程式中重複編寫 boto3 樣板程式碼。它還集中了驗證、回應正規化和工作階段處理，從而使整合行為保持一致。 設計上的工作階段連續性： 用戶端允許呼叫者傳入工作階段 ID 或建立一個新的工作階段 ID。這使得多輪工作流程變得明確，同時保留了簡單的單次（One-shot）呼叫。 * 正規化的回傳形狀： 回傳 {session_id, body} 使應用程式程式碼更容易測試，並避免了將底層 boto3 事件詳細資訊洩露給每個呼叫者。

實戰實驗室 C — 加入執行期錯誤分類法

開發者目標

為執行期用戶端和營運人員提供一致的錯誤類型，以應對缺少欄位、模型失敗和非預期異常。

開發者操作

建立 errors.py：

class RuntimeErrorCode:
    MISSING_PROMPT = "MISSING_PROMPT"
    VALIDATION_ERROR = "VALIDATION_ERROR"
    AGENT_FAILURE = "AGENT_FAILURE"
    UNEXPECTED_ERROR = "UNEXPECTED_ERROR"
def error_response(code: str, message: str, request_id: str | None = None) -> dict:
    return {
        "status": "error",
        "error": {
            "code": code,
            "message": message,
            "request_id": request_id,
        },
    }

修改執行期進入點模式：

from errors import RuntimeErrorCode, error_response
@app.entrypoint
def sovereign_runtime(payload, context):
    request_id = payload.get("request_id", context.session_id)
    try:
        prompt = payload.get("prompt", "")
        if not prompt.strip():
            return error_response(RuntimeErrorCode.MISSING_PROMPT, "提示詞 (prompt) 為必要欄位", request_id)
        response = agent(f"請求 ID: {request_id}\n工作階段: {context.session_id}\n{prompt}")
        return {"status": "ok", "request_id": request_id, "response": response.message["content"][0]["text"]}
    except Exception as exc:
        return error_response(RuntimeErrorCode.UNEXPECTED_ERROR, str(exc), request_id)

商業邏輯

營運系統需要可預測的錯誤，以便安全地進行路由、警示、重試或顯示。

程式碼邏輯

錯誤輔助程式回傳一個包含狀態、錯誤碼、訊息和請求 ID 的一致物件。

預期結果

格式錯誤的請求會回傳結構化錯誤，而不是不一致的字串或堆疊追蹤（Stack traces）。

系統設計決策

用於營運的錯誤分類法： 生產環境的用戶端需要區分驗證失敗與代理人失敗。基於代碼的分類法允許儀表板、重試邏輯和警示規則做出相應的反應。 每個錯誤中都包含請求 ID： 在錯誤中包含請求 ID 有助於將用戶端失敗與執行期日誌和工作階段 ID 相關聯。這減少了工作坊與生產事件期間的除錯時間。 * 安全的異常處理： 執行期擷取非預期錯誤並回傳一個受控的回應。這可以防止原始堆疊追蹤洩露給呼叫者，同時仍為開發者除錯保留足夠的詳細資訊。

實戰實驗室 D — 建構串流 CLI 消費者

開發者目標

建立一個用戶端，可以解析伺服器傳送事件（Server-Sent Event, SSE）風格的串流回應並逐步印出區塊。

開發者操作

建立 invoke_streaming_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 = str(uuid.uuid4())
response = client.invoke_agent_runtime(
    agentRuntimeArn=os.environ["AGENTCORE_STREAMING_RUNTIME_ARN"],
    runtimeSessionId=session_id,
    qualifier="DEFAULT",
    payload=json.dumps({
        "prompt": "串流分析信用利差擴大與主權重新定價訊號。"
    }).encode("utf-8"),
)
print("工作階段:", session_id)
for event in response["response"]:
    chunk = event.decode("utf-8") if isinstance(event, bytes) else str(event)
    print(chunk, end="", flush=True)
print()

商業邏輯

串流用戶端允許開發者查看漸進式的模型輸出，這對於聊天 UI 和分析師儀表板非常有用。

程式碼邏輯

指令碼呼叫串流執行期，並在回應事件到達時對其進行反覆運算。

預期結果

終端機會漸進式地印出內容，而不是等待完整回應。

系統設計決策

漸進式轉譯用戶端： 只有當用戶端正確使用區塊時，串流功能才有幫助。本實驗室教授的是串流的呼叫端，而不僅僅是執行期端。 獨立的串流執行期 ARN： 指令碼使用獨立的環境變數，以避免混淆同步與串流部署。這保持了測試的明確性。 * 立即排空輸出： flush=True 模擬了 UI 的漸進式轉譯，並有助於開發者在終端機中觀察串流行為。

實戰實驗室 E — 加入本地成本與延遲日誌包裝器

開發者目標

在應用程式層擷取模型呼叫延遲，以便進行工作坊除錯和未來的可觀測性建置。

開發者操作

建立 timed_agent.py：

import time
import json
from datetime import datetime, timezone
class TimedAgent:
    def __init__(self, agent, name: str):
        self.agent = agent
        self.name = name
    def __call__(self, prompt):
        start = time.time()
        response = self.agent(prompt)
        elapsed_ms = int((time.time() - start) * 1000)
        print(json.dumps({
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "event_type": "agent_invocation_latency",
            "agent": self.name,
            "elapsed_ms": elapsed_ms,
        }))
        return response

包裝代理人：

from timed_agent import TimedAgent
agent = TimedAgent(agent, "sovereign-runtime-agent")

商業邏輯

在新增受管的可觀測性儀表板之前，開發者需要對緩慢的模型呼叫和執行期行為具備可見性。

程式碼邏輯

TimedAgent 裝飾（Decorate）一個現有的 Strands 代理人，並為每次呼叫記錄耗費的毫秒數。

預期結果

每次呼叫都會印出一個 JSON 延遲事件。

系統設計決策

使用裝飾器而非侵入式修改： 包裝代理人可以避免更改代理人的建構或商業邏輯。這保持了可觀測性的模組化，且易於移除或替換。 結構化延遲日誌： JSON 延遲日誌可以被搜尋與彙整。它們還建立了與 AgentCore Observability 和 CloudWatch 的自然橋樑。 * 開發者回饋迴圈： 延遲資料有助於開發者在實戰實驗期間，了解較長提示詞、較大 Payload 以及額外工具所帶來的效能成本。