使用 AgentCore 與 Strands 建構：執行期蝕刻製程視窗測試自動化開發者工作坊

聽眾： 專業 Python/AWS 開發者

時長： 2 小時

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

工作坊形式： 開發者動手實作，無需簡報

專案產出： 一個具備本地驗證、部署指令碼、boto3 呼叫用戶端、串流變體、工作階段清理和大型承載資料（Payload）處理器的可用執行期代理程式（Runtime Agent）。

僅供教育工程工作坊使用。範例領域為蝕刻製程視窗（Etch-Process-Window）分析，但產出內容不構成製程控制、設備操作、安全、法律或合規性建議。

開發者將使用 Strands 與 AgentCore Runtime 建構一個生產級的蝕刻製程視窗代理程式。本工作坊涵蓋確定性計算工具、承載資料驗證、冒煙測試（Smoke Tests）、部署、boto3 呼叫、串流回應、大型 base64 承載資料處理以及工作階段清理。參與者將帶走可重複使用的執行期模式，以建構安全、可觀測、具備工作階段感知能力（Session-aware）的 AI 服務，並在 AWS 上將模型推理與可測試的 Python 邏輯相結合，以滿足現代企業開發者的工作流程。

1. 開發者學習目標

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

1. 建構一個由 Amazon Bedrock 模型支援的 Strands 代理程式。
2. 為 LLM 驅動的代理程式添加確定性 Python 工具（Tools）。
3. 將代理程式封裝在 Amazon Bedrock AgentCore Runtime 進入點中。
4. 使用 AgentCore 入門工具包（Starter Toolkit）打包並啟動執行期。
5. 使用 boto3 搭配穩定的工作階段 ID（Session ID）呼叫執行期。
6. 為高回應性的用戶端實作串流執行期進入點。
7. 建構一個支援 base64 編碼的 Excel 與圖片輸入的大型承載資料進入點。
8. 在部署前加入本地承載資料驗證與冒煙測試。

2. 本工作坊建構的架構

開發者筆記型電腦
  ├─ app.py                         # 同步 AgentCore 執行期進入點
  ├─ app_streaming.py               # 串流執行期進入點
  ├─ app_large_payload.py           # 多模態 / 大型承載資料執行期進入點
  ├─ deploy_runtime.py              # 打包並 launch 執行期
  ├─ invoke_runtime.py              # boto3 用戶端呼叫
  ├─ stop_session.py                # 明確的工作階段清理
  ├─ validate_payload.py            # 本地承載資料合約驗證
  ├─ 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-etch-runtime/prompts
cd agentcore-strands-etch-runtime
cat > requirements.txt <<'EOF'
bedrock-agentcore
bedrock-agentcore-starter-toolkit
strands-agents
strands-agents-tools
boto3
botocore
pytest
EOF

建立系統提示詞（System Prompt）：

cat > prompts/system.md <<'EOF'
你是一位專業的蝕刻製程視窗工程助理。
請分析良率漂移（Yield drifts）、CD-SEM 臨界尺寸漂移變寬、蝕刻製程視窗漂移、疊對漂移（Overlay drift）、
產能穩定性（Throughput stability）、等待時間需求（Queue-time demand）、批次流程（Lot flow）以及備用配方路由（Recipe fallback routing）。
請務必返回以下章節：
1. 觀測結果 (Observation)
2. 推理過程 (Reasoning)
3. 製程影響 (Process implication)
4. 控制考量 (Control 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_drift_bpu(fab_a_etch_rate: float, fab_b_etch_rate: float) -> str:
    """計算兩廠良率之間的基點單位（Basis units）漂移。"""
    drift = (fab_a_etch_rate - fab_b_etch_rate) * 100
    return f"良率漂移為 {drift:.1f} 基點單位。"

@tool
def overlay_control_count(wafer_count: float, control_ratio: float) -> str:
    """從晶圓（WAFERS）曝光量與控制比例計算控制名目值。"""
    control = wafer_count * control_ratio
    return f"疊對控制名目值為 WAFERS {control:,.2f}。"

@tool
def scrap_impact_notional(lot_count: float, drift_move_bpu: float, queue_time: float) -> str:
    """從漂移幅度估算由等待時間驅動的報廢影響名目值。"""
    loss = lot_count * queue_time * (drift_move_bpu / 10000)
    return f"預估的等待時間影響為 WAFERS {loss:,.2f}。"

agent = Agent(
    model=BedrockModel(
        model_id="amazon.nova-pro-v1:0",
        temperature=0.2,
        max_tokens=4000,
    ),
    tools=[calculator, yield_drift_bpu, overlay_control_count, scrap_impact_notional],
    system_prompt=SYSTEM_PROMPT,
)

@app.entrypoint
def etch_process_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 承載資料。

系統設計決策

• 定量邏輯使用確定性工具： LLM 適用於綜合分析，但工廠工程計算應具備確定性且可測試。透過將良率漂移、控制名目值和等待時間壓力數學邏輯移入 Python 工具，開發者可獲得可重複的輸出，並能獨立於模型行為之外對計算進行單元測試。
• 請求 ID 與工作階段內文： 進入點將請求與工作階段 ID 注入到模型提示詞中。這賦予了回應操作上的脈絡，並有助於工程師將執行期日誌、用戶端呼叫和使用者工作流程進行關聯。這也為生產環境的追蹤（Tracing）建立了乾淨的模式。
• 低溫度的 Bedrock 模型： 代理程式使用 temperature=0.2 以減少回應變異。專業的開發者工作坊應傳授穩定的工程模式，而非創意的提示詞實驗。較低的變異性有助於改善冒煙測試、展示以及未來的評估基準。

步驟 3 ── 新增本地承載資料驗證

開發者行動

建立 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": "分析 Fab-A/Fab-B 良率漂移", "request_id": "demo-001"}
    ok, errors = validate_payload(sample)
    print("是否有效:", ok)
    print("錯誤訊息:", errors)

商業邏輯

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

程式碼邏輯

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

預期結果

python validate_payload.py
# 是否有效: True
# 錯誤訊息: []

系統設計決策

• 部署前的承載資料合約： 開發者應在呼叫雲端執行期之前在本地驗證 API 合約。這減少了因缺少提示詞或拼錯欄位而導致的執行期呼叫失敗，也為日後建構更強健的 Schema 奠定了基礎。
• 明確的未知欄位偵測： 拒絕未知欄位有助於及早發現整合錯誤。若不這樣做，用戶端可能會誤以為中繼資料已被使用，而執行期其實默默忽略了它。清晰的驗證能提供開發者更好的回饋。
• 可重複使用的驗證模組： 驗證器是一個模組，而不僅僅是一個指令碼。它可以在單元測試、CLI 用戶端和執行期進入點中重複使用。這避免了在多個檔案中複製合約邏輯。

步驟 4 ── 新增冒煙測試

開發者行動

建立 smoke_test.py。

from validate_payload import validate_payload
from app import yield_drift_bpu, overlay_control_count, scrap_impact_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_drift_bpu(4.25, 2.15)
    assert "15,000,000.00" in overlay_control_count(25_000_000, 0.6)
    assert "125,000.00" in scrap_impact_notional(10_000_000, 25, 5)

執行：

pytest -q

商業邏輯

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

程式碼邏輯

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

預期結果

3 passed

系統設計決策

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

步驟 5 ── 部署 AgentCore 執行期

開發者行動

建立 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="etch-process-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"

商業邏輯

本地代理程式將成為一個託管的執行期端點（Endpoint），供應用程式呼叫。

程式碼邏輯

Runtime().configure() 打包進入點與依賴項。launch() 部署執行期並返回識別碼。

預期結果

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

系統設計決策

• 利用工具包提高動手實作速度： 入門工具包處理了部署機制，使工作坊能夠專注於執行期設計，而非容器內部細節。開發者仍能看到執行期 ARN，並可在事後檢查產生的 AWS 資源。
• 先部署單一執行期進入點： 僅部署 app.py 可保持首次雲端部署的簡單性。一旦同步呼叫正常運作，開發者就可以在減少未知變數的情況下部署串流或大型承載資料變體。
• 使用環境變數儲存 ARN： 將執行期 ARN 導出（Export）為環境變數，可將部署與呼叫指令碼解耦。這模擬了生產環境模式，其中部署輸出會作為用戶端設定或 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": (
        "分析 Fab-A 蝕刻率 4.25 與 Fab-B 蝕刻率 2.15。"
        "晶圓（WAFERS）曝光量為 25000000 且控制比例為 0.60。"
        "部位名目值為 10000000，漂移幅度為 25 bpu，等待時間（queue_time）為 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

商業邏輯

用戶端請求結構化的蝕刻製程視窗分析，並提供應觸發計算工具的數值。

程式碼邏輯

指令碼驗證承載資料、建立 boto3 AgentCore 用戶端、發送 JSON 位元組、結合執行期回應區塊（Chunks），並列印工作階段 ID。

預期結果

回應將包含分析內容以及計算出的數值，例如基點單位漂移、控制名目值或等待時間影響。

系統設計決策

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

步驟 7 ── 新增串流執行期變體

開發者行動

建立 app_streaming.py。

from strands import Agent, tool
from strands.models import BedrockModel
from bedrock_agentcore.runtime import BedrockAgentCoreApp

app = BedrockAgentCoreApp()

@tool
def metrology_drift_widening() -> str:
    """返回 CD-SEM 臨界尺寸漂移變寬的核心驅動因素。"""
    return "驅動因素包括等待時間壓力、產能撤回、良率損失風險和備用路由。"

agent = Agent(
    model=BedrockModel(model_id="amazon.nova-pro-v1:0", temperature=0.2, max_tokens=4000),
    tools=[metrology_drift_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 ── 新增大型承載資料處理器

開發者行動

建立 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=(
        "請分析工廠工程文件、SPC 圖表和製程圖片，以評估蝕刻製程視窗、良率漂移、臨界尺寸漂移、"
        "疊對漂移、產能、批次流程、控制考量、確認訊號與失效觸發條件。"
        "請勿提供製程發佈建議。"
    ),
)

@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": "etch_process_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 承載資料欄位進行解碼，並建構具備型別的 Bedrock 內容區塊：document、image 和 text。

預期結果

用戶端可以發送選填的 excel_data 和 image_data 欄位，並收到綜合分析報告。

系統設計決策

• 使用 Base64 進行 JSON 傳輸： 將二進位檔案進行編碼，使其可透過 JSON 承載資料傳輸。這保持了用戶端合約的簡單性，並避免了工作坊期間多部分表單上傳（Multipart upload）的複雜性。
• 具備型別的內容區塊： 傳入 document 與 image 區塊可保留模態性。這比將所有內容轉換為文字更具維護性，並能讓模型發揮特定文件/圖片的處理能力。
• 選填的承載資料欄位： 處理器可與純文字、純文件、純圖片或組合請求相容。這使得執行期對於多種類型的用戶端都具備彈性。

步驟 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"])

系統設計決策

• 工作階段是受控資源： 代理程式工作階段可以保留脈絡並消耗資源。開發者應在工作流程結束時明確停止工作階段，而非僅依賴閒置逾時（Idle expiration）。
• 清理指令碼化： 一個小型指令碼便於在展示、測試和 CI 工作任務結束後執行。這鼓勵了嚴謹的執行期環境維護衛生。
• 操作把手（Operational handle）： 工作階段 ID 成為偵錯、追蹤與清理的操作把手。

開發者最終檢查清單

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

附加開發者動手實作實驗室

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

動手實作實驗室 A ── 為執行期承載資料新增 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 檢查承載資料的外形、必要欄位、資料型別與未知欄位。該輔助程式會返回一個布林值以及易於閱讀的錯誤訊息。

預期結果

python validate_schema.py
# True
# []

系統設計決策

• Schema 作為共享的 API 合約： JSON Schema 比手寫驗證更精確，因為它在單一可重複使用的成品中定義了型別、必要欄位與未知欄位的行為。這允許用戶端、測試套件和執行期處理器在呼叫 AgentCore Runtime 之前驗證相同的合約。
• 在模型呼叫前快速失敗（Fail fast）： 執行期呼叫可能會產生模型延遲與費用。在呼叫代理程式之前驗證承載資料可以防止本可避免的失敗，並為開發者提供即時回饋。這也減少了由格式錯誤的用戶端承載資料引起的多餘執行期日誌。
• 可擴展的中繼資料（Metadata）物件： Schema 允許使用 metadata 物件來提供安全的操作脈絡，同時阻擋了任意的頂層欄位。這賦予團隊彈性，同時避免執行期合約變得失控。

動手實作實驗室 B ── 建構可重複使用的執行期用戶端類別

開發者目標

將 boto3 呼叫、工作階段 ID、承載資料驗證和回應標準化封裝到一個可重複使用的用戶端類別中。

開發者行動

建立 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="分析 Fab-A/Fab-B 良率漂移幅度，並包含確認與失效訊號。",
    request_id="client-lab-001",
    metadata={"application": "developer-lab"},
)
print(result["session_id"])
print(result["body"])

商業邏輯

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

程式碼邏輯

該類別驗證承載資料、建立或重用工作階段 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 etch_process_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）。

系統設計決策

• 用於維運的錯誤分類法： 生產環境的用戶端需要區分驗證失敗與代理程式失敗。基於代碼（Code-based）的分類法允許儀表板、重試邏輯和告警規則做出適當的反應。
• 每個錯誤中皆包含請求 ID： 在錯誤中包含請求 ID 有助於將用戶端失敗與執行期日誌、工作階段 ID 進行關聯。這減少了工作坊與生產突發事件期間的偵錯時間。
• 安全的異常處理： 執行期捕獲非預期的錯誤並返回一個有界的（Bounded）回應。這可防止原始堆疊追蹤洩露給呼叫者，同時仍保留足夠的詳細資訊供開發者偵錯。

動手實作實驗室 D ── 建構串流 CLI 消費者

開發者目標

建立一個用戶端，能夠解析伺服器傳送事件（Server-Sent Events, 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": "請串流分析 CD-SEM 臨界尺寸漂移變寬與蝕刻製程漂移訊號。"
    }).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 和工程師儀表板非常有用。

程式碼邏輯

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

預期結果

終端機會逐步列印內容，而不是等待完整回應生成完畢才顯示。

系統設計決策

• 漸進式渲染用戶端： 串流只有在用戶端正確消費區塊時才有幫助。本實驗室教授的是串流的呼叫端，而不僅僅是執行期端。
• 獨立的串流執行期 ARN： 指令碼使用獨立的環境變數，以避免混淆同步與串流部署。這保持了測試的明確性。
• 立即清除快取輸出（Flush）： 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, "etch-process-runtime-agent")

商業邏輯

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

程式碼邏輯

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

預期結果

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

系統設計決策

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