【Day 12】系統全監控：/health 與 /stats API 實作與 Ansible 巡檢整合

2026 iThome 鐵人賽

DAY 12

Build on Google AI

打造零成本企業級 AI Agent：以 Gemini 2.5 Flash 構建金融分析助手與維運實戰系列第 12 篇

18th鐵人賽 gemini ai agent fastapi monitoring

ariel_c

2026-08-14 12:28:04

59 瀏覽

分享至

「一個真正符合 Production Ready 標準的自建 AI Agent，除了具備完善的業務邏輯，更需要具備可視化的系統健康指標與自動化巡檢能力。」

在 Day 11 中，我們探討了前端靜態檔案掛載與 static/validation.js 前後端縱深防禦機制。今天我們將深入 app/main.py 的核心監控功能——/health 與 /stats 指標 API 的實作，以及如何與 Day 03 撰寫的 Ansible 自動化巡檢劇本 (status.yml) 進行深度整合！

透過這套監控機制，維運者無須登入伺服器，即可隨時掌握 Agent 的運行狀態、對話輪數與向量知識庫的即時容量。

本篇重點摘要

剖析 app/main.py 中 /health（健康檢查）與 /stats（系統統計）API 的設計細節。
整合 ConversationMemory 與 RAGEngine 的即時數據查詢（Turn Count & Vector Count）。
解析 structlog JSON 格式日誌寫入與 /var/log/angelina/app.log 備援機制。
連結 Ansible status.yml 自動化巡檢劇本，實現一鍵式無人值守維運。

一、系統健康檢查與指標 API 實作 (app/main.py)

在 FastAPI 中，我們設計了兩個輕量且高效的 Endpoint，提供負載均衡器（Load Balancer）、Ansible 或外部 Monitoring Agent 隨時探測：
1. 輕量健康檢查 (GET /health)
用於最基礎的 HTTP 存活探針（Liveness Probe），回傳 HTTP 200 與 {"status": "ok"}：

Python
@app.get("/health")
async def health_check() -> dict:
"""Health check endpoint. Returns HTTP 200 with status ok."""
return {"status": "ok"}

  2. 系統狀態與指標統計 (GET /stats)
  呼叫 ConversationMemory 與 RAGEngine 的單例服務（Singletons），非同步讀取當前 SQLite 中的對話記錄總數與 ChromaDB 向量庫中的文檔總數：

Python
@app.get("/stats")
async def get_stats() -> dict:
    """Return system statistics: memory turn count and vector count."""
    assert _memory is not None
    assert _rag_engine is not None

    memory_turns = await _memory.get_turn_count()
    vector_count = await _rag_engine.get_collection_count()

    return {"memory_turns": memory_turns, "vector_count": vector_count}

二、結構化日誌紀錄（structlog）與權限容錯設計

為了支援自動化日誌分析工具（如 ELK 或 Promtail/Loki），系統全程使用 structlog 輸出標準 ISO 8601 時間戳與 JSON 格式日誌。

在 _configure_logging() 函式中，我們還設計了權限降級（Fallback）邏輯：當系統無權限寫入 /var/log/angelina/app.log 時（例如在開發環境或非 Root 容器中執行），自動降級輸出至 sys.stderr，確保應用程式絕不因日誌檔建立失敗而崩潰：

Python
def _configure_logging() -> None:
    """Configure structlog with JSON output to /var/log/angelina/app.log."""
    log_dir = Path("/var/log/angelina")
    log_file_path = log_dir / "app.log"

    try:
        log_dir.mkdir(parents=True, exist_ok=True)
        log_file = open(log_file_path, "a", encoding="utf-8")
    except (PermissionError, OSError):
        # 備援機制：權限不足時自動切換至 stderr
        log_file = sys.stderr

    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.StackInfoRenderer(),
            structlog.processors.format_exc_info,
            structlog.processors.JSONRenderer(),
        ],
        wrapper_class=structlog.make_filtering_bound_logger(0),
        logger_factory=structlog.PrintLoggerFactory(file=log_file),
        cache_logger_on_first_use=True,
    )

三、Ansible 自動化巡檢劇本整合 (ansible/status.yml)

有了 /health 與 /stats API 後，我們便能將它們與 Day 03 所建構的 Ansible IaC 運維體系結合。

維運人員只需在管理端執行 ansible-playbook -i inventory.ini status.yml，Ansible 便會對 RHEL 宿主機發起自動化巡檢：

YAML
# ansible/status.yml 巡檢邏輯範例
- name: Inspect Angelina Agent Status
  hosts: angelina_servers
  tasks:
    - name: Check systemd service status
      systemd:
        name: angelina
      register: service_info

    - name: Call /health endpoint
      uri:
        url: "http://127.0.0.1:8080/health"
        method: GET
        status_code: 200
      register: health_res

    - name: Call /stats endpoint
      uri:
        url: "http://127.0.0.1:8080/stats"
        method: GET
      register: stats_res

    - name: Display Agent Monitoring Summary
      debug:
        msg:
          - "Service Status: {{ service_info.status.ActiveState }}"
          - "Health Status: {{ health_res.json.status }}"
          - "Total Memory Turns: {{ stats_res.json.memory_turns }}"
          - "Total Vector Knowledge: {{ stats_res.json.vector_count }}"