把楼宇诊断从 20 分钟压缩到 20 秒：基于 Amazon Bedrock AgentCore 的智能体架构

1200 / 20 = 60

自然语言界面
    ↓
AgentCore 中的智能体运行与会话控制
    ↓
受约束的诊断工具：资产、遥测、告警、工单、知识库
    ↓
楼宇管理系统、时序数据库、数据湖和维护系统

fastapi==0.115.0
uvicorn[standard]==0.30.6
pydantic==2.9.2

from time import perf_counter
from typing import Literal

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, Field

app = FastAPI(title="Building Diagnostic Tools")

# 示例数据；生产环境中应替换为实时遥测或时序数据库查询。
BUILDINGS = {
    "BLDG-001": {
        "zones": {
            "north": {"temperature_c": 25.8, "setpoint_c": 22.0},
            "south": {"temperature_c": 22.4, "setpoint_c": 22.0},
        },
        "ahu": {
            "supply_air_c": 17.5,
            "static_pressure_pa": 145,
            "expected_pressure_pa": 200,
        },
        "active_alarms": ["AHU-1 low static pressure"],
    }
}


class DiagnosticRequest(BaseModel):
    building_id: str = Field(examples=["BLDG-001"])
    question: str = Field(
        examples=["Why is the north zone warmer than its setpoint?"]
    )


class Evidence(BaseModel):
    source: str
    metric: str
    actual: float | str
    expected: float | str


class DiagnosticResponse(BaseModel):
    status: Literal["normal", "attention_required"]
    summary: str
    recommended_actions: list[str]
    evidence: list[Evidence]
    elapsed_ms: float


@app.post("/tools/diagnose", response_model=DiagnosticResponse)
def diagnose(
    request: DiagnosticRequest,
    x_role: str = Header(default="viewer"),
):
    started = perf_counter()

    if x_role not in {"operator", "engineer"}:
        raise HTTPException(status_code=403, detail="Insufficient role")

    building = BUILDINGS.get(request.building_id)
    if building is None:
        raise HTTPException(status_code=404, detail="Building not found")

    north = building["zones"]["north"]
    ahu = building["ahu"]
    temperature_delta = north["temperature_c"] - north["setpoint_c"]
    pressure_low = ahu["static_pressure_pa"] < ahu["expected_pressure_pa"]

    evidence = [
        Evidence(
            source="zone_sensor:north",
            metric="temperature_c",
            actual=north["temperature_c"],
            expected=north["setpoint_c"],
        ),
        Evidence(
            source="ahu:AHU-1",
            metric="static_pressure_pa",
            actual=ahu["static_pressure_pa"],
            expected=ahu["expected_pressure_pa"],
        ),
    ]

    if temperature_delta > 2 and pressure_low:
        status = "attention_required"
        summary = (
            "The north zone is 3.8°C above setpoint while AHU-1 static "
            "pressure is below its expected value. Insufficient airflow is "
            "a plausible cause, but it must be verified on site."
        )
        actions = [
            "Inspect AHU-1 filters and fan operation.",
            "Check north-zone dampers for blockage or control faults.",
            "Review the last 24 hours of pressure and temperature trends.",
        ]
    else:
        status = "normal"
        summary = "No correlated temperature and airflow anomaly was found."
        actions = ["Continue monitoring current telemetry."]

    elapsed_ms = round((perf_counter() - started) * 1000, 2)
    return DiagnosticResponse(
        status=status,
        summary=summary,
        recommended_actions=actions,
        evidence=evidence,
        elapsed_ms=elapsed_ms,
    )

python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
uvicorn app:app --host 0.0.0.0 --port 8000

curl -s http://localhost:8000/tools/diagnose \
  -H 'Content-Type: application/json' \
  -H 'X-Role: engineer' \
  -d '{
    "building_id": "BLDG-001",
    "question": "Why is the north zone warmer than its setpoint?"
  }' | python -m json.tool

You are a building diagnostic assistant.
Use only data returned by approved tools.
Separate measured facts from diagnostic hypotheses.
Cite the source and metric for every anomaly.
Never claim that equipment has failed unless a tool explicitly confirms it.
For write operations, request human approval before calling the tool.