JamesIOmete/iot-ops-agent

# iot-ops-agent 一个用于物联网冷链车队运营的自主 AI agent。利用 Anthropic Claude API 及其工具使用功能来监控设备健康状况、调查告警、执行 Runbook 并进行升级上报——全程无需人工干预。 旨在展示高级架构师如何思考 Agentic 系统：不仅关注其能否运行，更关注其构建方式背后的原因。每一个设计决策都经过深思熟虑并留有文档记录。 ## 功能简介 三种操作模式，单一 agent 循环： | 模式 | 触发条件 | 执行内容 | |------|---------|--------------| | **Watchdog** | 计划执行（例如每 15 分钟） | 完整的车队健康检查。检查所有设备、所有告警、DLQ 深度。生成结构化报告。 | | **Incident** | CloudWatch 触发告警 | 执行 YAML 格式的 Runbook。收集证据、形成假设、评估置信度、执行修复或进行升级上报。 | | **Briefing** | 按需执行（运维通话，轮班交接） | 为值班工程师提供简明的车队状态报告。按严重程度排序。 | ## 运行方式 ``` # 安装依赖 python -m venv .venv && source .venv/bin/activate pip install -r requirements.txt # mock 模式下的 Watchdog (无需 AWS 凭证) ANTHROPIC_API_KEY=sk-... python run.py --mode watchdog # 带有特定 runbook 的 Incident responder ANTHROPIC_API_KEY=sk-... python run.py --mode incident --runbook temperature_excursion # 启用了写入操作的 Incident (创建 GitHub issues, 发布 SNS) ANTHROPIC_API_KEY=sk-... python run.py --mode incident --runbook device_silence \ --allow-write --no-mock # 按需简报 ANTHROPIC_API_KEY=sk-... python run.py --mode briefing ``` ### 命令行参数 | 参数 | 默认值 | 描述 | |------|---------|-------------| | `--mode` | 必填 | `watchdog`、`incident` 或 `briefing` | | `--no-mock` | 关闭（mock 开启） | 使用真实的 AWS 而非 fixture 数据 | | `--allow-write` | 关闭 | 启用 `create_github_issue` 和 `publish_sns_escalation` | | `--config-override PATH` | 无 | 经深度合并覆盖 `config/config.yaml` 的 YAML 文件 | | `--runbook ID` | `temperature_excursion` | 要执行的 Runbook（仅限 incident 模式） | | `--device ID` | 无 | 在 incident 模式下定位特定设备 | ## 架构简述 ``` flowchart LR run["run.py"] --> core["AgentCore"] core --> td["tool dispatch"] subgraph aws ["AWS"] dynamo["DynamoDB"] cw["CloudWatch"] cwl["CloudWatch Logs"] sqs["SQS"] shadow["IoT Core Shadow"] end td --> dynamo td --> cw td --> cwl td --> sqs td --> shadow td -. "mock mode" .-> fix[("tests/fixtures/")] core --> logs[("JSONL logs")] core --> esc["escalation.py"] esc --> gate{"--allow-write?"} gate -->|enabled| sns["SNS"] gate -->|enabled| gh["GitHub Issues"] ``` agent 循环是一个标准的工具使用循环：调用 Claude，分发所有工具调用，将结果添加到上下文中，重复此过程直到 `end_turn`。使其具备生产就绪能力的关键在于环绕该循环的周边设计： - **双层权限执行** — `AgentCore._dispatch_tool()` 在任何工具调用之前会检查模式允许列表和写入标志。写入工具会独立检查相同的标志。这是纵深防御，而不是冗余设计。 - **增量结构化日志** — 每个事件（工具调用、工具结果、推理步骤、升级上报）都会作为 JSONL 立即写入并刷新。在运行期间发生崩溃也会留下有效的部分日志。 - **升级上报作为一等结果** — `escalation.py` 包含所有升级逻辑。置信度低于阈值或建议操作中出现受阻止的关键词都会触发升级上报。模式文件从不做出此决定。 - **配置作为单一事实来源** — 所有阈值、工具权限和 AWS 资源名称都存在于 `config/config.yaml` 中。agent 代码中没有魔法数字。 - **默认为 Mock 模式** — 该 agent 可以在没有 AWS 凭证的情况下完全运行。`tests/fixtures/` 中的 fixture 数据讲述了一个连贯的故事：truck-002 出现了温度超标，truck-003 处于失联状态。 有关完整的设计原理，请参阅 [`docs/architecture.md`](docs/architecture.md)。 ## 设计决策 有边界的工具使用：agent 必须具有显式的、声明式的工具注册表并在分发时强制执行，而不是赋予 LLM 无限制的能力。在生产环境中，使用不恰当或未经批准的工具的 agent 可能会导致错误的数据更改、意料之外的安全隐患、成本增加或其他面向客户的不良影响。 升级上报：我将升级上报视为一种经过设计的控制路径，而不是故障处理。agent 应当知道何时其置信度、策略、面临歧义或风险超出了其权限范围，并生成带有结构化上下文的平滑交接。如果没有该路径，生产环境中的 agent 可能会进行猜测、停滞、盲目重试，或者将重要的不确定性丢失在通用的错误处理中。 结构化推理日志：我将所有 agent 活动记录为结构化的 JSONL 事件。生产环境中的行为需要在决策、工具调用、成本、错误和结果之间具有可检查、可重放、可搜索和可追溯的特性。如果没有结构化日志，调试将变得混乱，事件回顾将沦为猜测，并且可能导致错失运维经验。 默认为 Mock 模式：我发布该 agent 时默认启用 `mock_mode: true`，并要求在接触真实 AWS 资源之前提供显式的 `--no-mock` 标志。现实世界中的副作用应需要明确的意图，尤其是在作品集代码、演示和早期开发循环中。这可以保持迭代的的安全性，防止意外的云更改或花费，并体现了在真实客户环境中所期望的生产纪律。 ## 工具注册表 跨越三个权限层级的 9 个工具。大多数模式下都可以使用读取工具；写入工具需要显式使用 `--allow-write`： | 工具 | Watchdog | Incident | Briefing | AWS 服务 | |------|:--------:|:--------:|:--------:|-------------| | `get_device_telemetry` | ✓ | ✓ | ✓ | DynamoDB | | `list_fleet_devices` | ✓ | ✓ | ✓ | DynamoDB | | `get_excursion_events` | ✓ | ✓ | ✓ | DynamoDB | | `get_cloudwatch_alarm_state` | ✓ | ✓ | ✓ | CloudWatch | | `query_cloudwatch_logs` | ✓ | ✓ | ✗ | CloudWatch Logs | | `get_dlq_depth` | ✓ | ✓ | ✓ | SQS | | `get_device_shadow` | ✗ | ✓ | ✗ | IoT Core Shadow | | `create_github_issue` | ✗ | ✓* | ✗ | GitHub API | | `publish_sns_escalation` | ✗ | ✓* | ✗ | SNS | `✓*` 需要 `--allow-write`。 完整的 Schema 和 fixture 详情：[`docs/tool-registry.md`](docs/tool-registry.md) ## Runbook Runbook 是位于 `config/runbooks/` 中的声明式 YAML 文件。它们定义了步骤、工具提示、假设和置信度调整。agent 将它们作为上下文读取——它会进行解释，而不是逐行执行。这意味着 agent 可以跳过不相关的步骤、合并步骤，或者呈现 Runbook 未曾预料到的证据。 包含两个 Runbook： - `temperature_excursion.yaml` — 5 步调查：识别受影响设备 → 超标历史记录 → 告警验证 → 设备 Shadow → 日志分析 - `device_silence.yaml` — 4 步调查：识别失联设备 → 设备 Shadow → DLQ 检查 → 日志分析 ## 测试 ``` pytest tests/ -v # 20 个测试，无需 AWS 凭证，约 0.3 秒 ``` 每个工具在 mock 模式下至少有一个测试。Fixture 数据非常逼真：真实的温度范围、真实的设备 ID，以及涵盖所有 fixtures 的连贯故障场景。`tests/test_tools.py` 是 CI 的基线。 ## 推理日志 每次运行都会在 `logs/` 目录下生成一个 JSONL 文件。事件如下： ``` run_start → llm_request → llm_response → tool_call → tool_result → ... → run_end ``` 示例 `run_end` 事件： ``` { "type": "run_end", "status": "complete", "total_input_tokens": 24381, "total_output_tokens": 847, "tool_calls_made": 14, "log_path": "logs/watchdog_20260506T143012Z_a3f8b21c.jsonl", "timestamp": "2026-05-06T14:31:48.221Z", "run_id": "a3f8b21c" } ``` 每个工具结果中的 `data_source` 字段显示为 `"mock"` 或实际的 AWS 服务名称——这使得日志中每个数据点的来源都清晰可见。 ## 技术栈 - Python 3.11+ - [Anthropic Python SDK](https://github.com/anthropics/anthropic-sdk-python) — 使用工具调用的 `claude-sonnet-4-6` - boto3 — DynamoDB, CloudWatch, CloudWatch Logs, SQS, SNS, IoT Core - PyGithub — GitHub issue 创建 - PyYAML — 配置和 Runbook 加载 - pytest — 测试套件 ## 相关项目 - **[aws-iot-edge-reference](https://github.com/JamesIOmete/aws-iot-edge-reference)** — 该 agent 监控的 AWS IoT 技术栈：Greengrass、IoT Core 规则、DynamoDB schema、CloudWatch 告警 - **[iotctl](https://github.com/JamesIOmete/iotctl)** — 用于车队管理操作的 CLI（固件更新、Shadow 检查、批量命令） - **[tf-plan-ai-reviewer](https://github.com/JamesIOmete/tf-plan-ai-reviewer)** — 针对该技术栈底层基础设施的 AI 辅助 Terraform plan 审查工具 - **[multicloud-sa-toolkit](https://github.com/JamesIOmete/multicloud-sa-toolkit)** — 跨 AWS、Azure 和 GCP 的解决方案架构师参考模式

标签：Anthropic Claude, AWS, CISA项目, CloudWatch, DLL 劫持, DPI, Python, SNS, SRE, 偏差过滤, 冷链物流, 大语言模型, 安全规则引擎, 恶意代码分类, 报警处理, 无人工干预, 无后门, 架构设计, 物联网, 自动化运行手册, 设备健康管理, 车队管理, 运维自动化