如何设计 AI 驱动的 OPS Agent：漫谈状态机

flowchart LR
  subgraph UI/API
    H[Human/UI]
  end

  subgraph Core[Core Services]
    O[Orchestrator\nFSM+Outbox+Timeline]
    P[Planner\n(LLM/模板→ plan_proposed)]
    G[Gatekeeper\n(策略/OA判定)]
    E[Executor\n(Adapters: script/k8s/...)]
    V[Verifier\n(SLO阈值)]
    A[Analyst\n(规则/统计→ findings)]
    S[Sensor\n(oo_locator / ingest)]
    L[Librian\n(pgvector RAG)]
  end

  EV[(Event Bus\nNATS/Redpanda)]
  DB[(PostgreSQL/Timescale)]
  OO[(OpenObserve/Logs)]

  H -- REST --> O
  O -- outbox/cmd.* --> EV
  EV -- evt.* --> O
  P & G & E & V & A & S & L -- evt./cmd. <--> EV
  O -- R/W --> DB
  E -- logs/refs --> OO

- REST：
  - `POST /case/create`，`GET /case/{id}`，`GET /case/{id}/timeline`
  - `PATCH /case/{id}/transition`（**唯一改状态入口**；需 `Idempotency-Key` + 建议 `If-Match`）
  - `POST /plan/generate`，`POST /gate/eval`，`POST /adapter/exec`，`POST /verify/run`
- 事件：
  - `evt.case.transition.v1`（任何迁移事实）
  - `evt.plan.proposed.v1`，`evt.plan.approved.v1`
  - `evt.exec.step_result.v1`，`evt.exec.done|failed.v1`
  - `evt.verify.pass|failed.v1`，`evt.analysis.findings.v1`
  - `cmd.plan.generate / cmd.gate.eval / cmd.exec.run / cmd.verify.run`
---

stateDiagram-v2
  [*] --> NEW
  NEW --> ANALYZING: start_analysis
  ANALYZING --> PLANNING: analysis_done
  PLANNING --> WAIT_GATE: plan_ready (guard: complete plan)
  WAIT_GATE --> EXECUTING: gate_approved
  WAIT_GATE --> PARKED: gate_rejected
  EXECUTING --> VERIFYING: exec_done
  EXECUTING --> PARKED: exec_failed
  VERIFYING --> CLOSED: verify_pass
  VERIFYING --> PARKED: verify_failed
  PARKED --> PLANNING: resume / fix

func Transition(ctx, caseID, event, meta) (Case, error) {
  return WithTx(ctx, func(tx Tx) (Case, error) {
    c := repo.GetForUpdate(tx, caseID)
    allowed, next := workflow.Decide(c.Status, event)
    if !allowed { return error(409) }

    // guards (schema/plan completeness/risk window etc.)
    if err := Guards.Pass(tx, c, event, meta); err != nil { return error(422) }

    c.Status, c.Version = next, c.Version+1
    repo.Update(tx, c)

    timeline.Append(tx, c.ID, event, meta)

    outbox.Append(tx, BuildMessages(c, event, meta))

    return c, nil
  })
}

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "title": "ChangePlan",
  "type": "object",
  "required": ["title","steps","rollback","verify"],
  "properties": {
    "title": {"type":"string"},
    "risk_score": {"type":"number","minimum":0,"maximum":1},
    "steps": {
      "type":"array","minItems":1,
      "items": {"type":"object","required":["action","timeout_s","retry"],
        "properties": {
          "action": {"enum":["k8s.rollout","script.exec","gateway.flip","noop"]},
          "params": {"type":"object"},
          "timeout_s": {"type":"integer","minimum":10},
          "retry": {"type":"integer","minimum":0,"maximum":3}
        }
      }
    },
    "rollback": {"type":"object"},
    "verify": {"type":"object"}
  }
}

System: 你是资深 SRE。产出必须是符合 JSON Schema 的 ChangePlan。不要输出自然语言解释。
User: {问题摘要/证据/限制}
Assistant: {ChangePlan JSON}

-- 1) 事实源
CREATE TABLE ops_case (
  id UUID PRIMARY KEY,
  tenant TEXT, title TEXT,
  status TEXT NOT NULL,
  version INT NOT NULL,
  created_at TIMESTAMPTZ DEFAULT now(),
  updated_at TIMESTAMPTZ DEFAULT now()
);

CREATE TABLE case_timeline (
  id BIGSERIAL PRIMARY KEY,
  case_id UUID REFERENCES ops_case(id),
  event TEXT, actor TEXT, reason TEXT,
  correlation_id TEXT, meta JSONB,
  created_at TIMESTAMPTZ DEFAULT now()
);

CREATE TABLE outbox (
  id BIGSERIAL PRIMARY KEY,
  topic TEXT, key TEXT, payload JSONB,
  created_at TIMESTAMPTZ DEFAULT now(),
  published_at TIMESTAMPTZ
);

CREATE TABLE idempotency (
  key TEXT PRIMARY KEY,
  response JSONB,
  created_at TIMESTAMPTZ DEFAULT now()
);

-- 2) Planner/Exec/Verify（示意）
CREATE TABLE plan_proposed (
  id UUID PRIMARY KEY,
  case_id UUID REFERENCES ops_case(id),
  plan JSONB NOT NULL,
  created_at TIMESTAMPTZ DEFAULT now()
);
CREATE TABLE plan_approved (
  id UUID PRIMARY KEY,
  case_id UUID REFERENCES ops_case(id),
  decision JSONB,
  created_at TIMESTAMPTZ DEFAULT now()
);

# 1) 创建 Case
CASE=$(curl -sX POST /case/create -d '{"title":"p95 spike","tenant":"t-001"}'|jq -r .data.case_id)

# 2) 开始分析
curl -X PATCH /case/$CASE/transition \
  -H 'Idempotency-Key:a1' \
  -d '{"event":"start_analysis","actor":"analyst@svc"}'

# 3) 生成计划（模板/LLM）
curl -X POST /plan/generate -d '{"case_id":"'$CASE'","template":"k8s.rollout.canary"}'
# Planner 发 evt.plan.proposed.v1 → Orchestrator 判定 plan_ready → WAIT_GATE

# 4) 自动审批
curl -X POST /gate/eval -d '{"case_id":"'$CASE'"}'
# Gatekeeper 发 evt.plan.approved.v1 → EXECUTING

# 5) 执行适配器（示例：script）
curl -X POST /adapter/exec -d '{"case_id":"'$CASE'","adapter":"script","script":"echo ok"}'
# Executor 发 evt.exec.done.v1 → VERIFYING

# 6) 验证通过 → CLOSED
curl -X POST /verify/run -d '{"case_id":"'$CASE'"}'

.
├─ api/                # Gin/Fiber + OpenAPI + 中间件
├─ workflow/           # 纯函数 FSM
├─ plans/              # 纯函数计划生成（模板 + LLM 契约）
├─ gatekeeper/         # 策略引擎（本地规则/OPA/Cedar）
├─ executor/
│  ├─ adapters/        # k8s/script/gateway...
│  └─ runner/          # step 执行/超时/重试
├─ verifier/
├─ analyst/
├─ sensor/
├─ librarian/
├─ internal/pubsub/    # NATS/Redpanda（至少一次 + 去重）
├─ ports/              # Repo/Outbox/Timeline/Idempotency 接口
├─ services/           # Orchestrator 等聚合服务
└─ migrations/         # DDL