使用 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. 開發者學習目標

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

● 建構一個由 Amazon Bedrock 模型支援的 Strands 代理人。

● 為以 LLM 驅動的代理人添加決定性的 Python 工具。

● 將代理人封裝至 Amazon Bedrock AgentCore Runtime 進入點（Entrypoint）。

● 使用 AgentCore 入門工具包（Starter Toolkit）封裝並啟動執行期。

● 使用穩定的工作階段 ID（Session ID）透過 boto3 呼叫執行期。

● 為需要即時回應的用戶端實作串流執行期進入點。

● 建構一個可處理 base64 編碼的 Excel 和圖片輸入的大容量 Payload 進入點。

● 在部署前加入本地 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 小時實戰議程

---

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 以及額外工具所帶來的效能成本。