NikhilRaman12/Incident-Renponse-Env
GitHub: NikhilRaman12/Incident-Renponse-Env
这是一个基于OpenEnv标准的SRE事件响应仿真环境，旨在通过模拟真实生产警报与指标，评估AI智能体在故障分类、根因分析及Runbook生成等任务中的表现。
Stars: 0 | Forks: 0
# 🚨 IncidentResponseEnv
[](https://openenv.ai)
[](https://huggingface.co/spaces/openenv/incident-response-env)
[](#testing)
[](https://python.org)
**一个模拟真实世界 Site Reliability Engineering (SRE) 事件响应的生产级 OpenEnv 环境。** AI 代理必须对生产警报进行分流，从日志和指标中进行根本原因分析，并生成可执行的 Runbook——这正是每家大型科技公司值班工程师使用的确切工作流程。
## 为什么选择这个领域？
每个运行生产系统的软件公司都会面临同样的凌晨 3 点噩梦：警报触发、指标飙升，人类工程师只有几分钟的时间来诊断和解决问题，否则用户将受到影响。这是 AI 能够有意义地自动化或协助的最**高风险**、**知识密集型**、**真实任务**之一。与玩具环境不同，IncidentResponseEnv 中的每个操作都具有与真实事件相同的后果结构：
- 将 P1 误分类为 P4 = 错过关键事件
- 错误的根本原因分析 = 错误的修复，更多的停机时间
- 不完整的 Runbook = 工程师在混乱中盲目行动
## 🗂️ 环境结构
```
incident-response-env/
├── openenv.yaml # OpenEnv spec metadata
├── Dockerfile # Container deployment
├── requirements.txt
├── app.py # FastAPI app (HF Spaces + REST API)
├── environment/
│ ├── __init__.py
│ ├── models.py # Typed Pydantic v2 models
│ ├── env.py # Core OpenEnv: step/reset/state
│ ├── graders.py # Deterministic graders (all 3 tasks)
│ └── data_generator.py # Realistic scenario synthesis
├── baseline/
│ └── run_baseline.py # OpenAI API baseline evaluation
└── tests/
└── test_env.py # 15 validation tests
```
## 🔌 OpenEnv API
### `reset(seed?) → Observation`
开始一个新回合。返回包含实时事件警报、指标、日志和服务上下文的初始观察。
```
from environment.env import make_env
env = make_env(task_type="alert_classification", seed=42)
obs = env.reset()
print(obs.alert.title) # "CRITICAL: Database connection pool exhausted"
print(obs.alert.metrics) # {"db_connection_pool_utilization_pct": 100.0, ...}
print(obs.alert.logs) # ["ERROR: sorry, too many clients already", ...]
print(obs.to_agent_prompt()) # Full formatted prompt for LLM
```
### `step(action) → (Observation, Reward, done, info)`
处理一个代理操作。返回下一个观察、带有细分的评分 Reward、完成标志和诊断信息。
```
from environment.models import Action, ActionType, SeverityLevel, AlertCategory
action = Action(
action_type = ActionType.CLASSIFY_ALERT,
severity = SeverityLevel.P1,
category = AlertCategory.DATABASE,
team = "database-team",
notes = "DB connection pool at 100%, scheduler bulk job is root cause",
)
obs, reward, done, info = env.step(action)
print(reward.score) # 0.9200
print(reward.breakdown) # RewardBreakdown(severity_accuracy=0.4, ...)
print(reward.feedback) # "✓ Severity P1 is correct. ✓ Category 'database' is correct..."
```
### `state() → EpisodeState`
返回评估框架所需的完整回合状态。
```
state = env.state()
print(state.step) # 1
print(state.cumulative_reward) # 0.92
print(state.actions_taken) # [{"step": 1, "action_type": "classify_alert", ...}]
```
## 📋 任务
### 任务 1 — 警报分类 `[简单]` 🟢
**代理必须做什么：** 给定一个带有指标和日志片段的实时生产警报，进行分类：
- **严重性**：P1（严重）/ P2（高）/ P3（中）/ P4（低）
- **类别**：基础设施 / 应用程序 / 数据库 / 网络 / 安全
- **团队**：需要通知哪个值班团队
**为什么它很现实：** 这正是事件响应的前 90 秒。值班工程师被叫醒，看到警报，必须立即做出决定：“这有多糟糕？谁需要知道？”
**奖励细分：**
| 组件 | 权重 | 描述 |
|-----------|--------|-------------|
| `severity_accuracy` | 0.40 | 完全匹配 = 1.0；相差一级 = 0.5；相差很大 = 0.0 |
| `category_accuracy` | 0.30 | 完全匹配 = 1.0；错误 = 0.0 |
| `team_accuracy` | 0.20 | 团队名称的模糊匹配 |
| `completeness_bonus` | 0.10 | 所有字段 + 备注已填写 |
| `penalty_misclass` | −0.20 | P1 被分类为 P3/P4（关键的分流不足） |
**最大步数：** 5
### 任务 2 — 根本原因分析 `[中等]` 🟡
**代理必须做什么：** 给定已分类的警报以及扩展的日志、指标时间序列和服务依赖图，识别：
- **根本原因组件**（哪个服务/组件发生了故障）
- **根本原因类型**（connection_pool_exhaustion、memory_leak 等）
- **证据**（3-5 个支持诊断的具体日志/指标观察）
- **影响**（爆炸半径和面向用户的后果）
**为什么它很现实：** 调查阶段需要关联多个信号——部署时间戳、可疑日志行、超过阈值的指标——并将它们综合成因果解释。
**奖励细分：**
| 组件 | 权重 | 描述 |
|-----------|--------|-------------|
| `rca_component` | 0.25 | 根本原因组件名称的模糊匹配 |
| `rca_type` | 0.25 | 故障分类的模糊匹配 |
| `evidence_quality` | 0.25 | 与真实证据的关键词重叠 |
| `impact_accuracy` | 0.15 | 影响关键词的覆盖率 |
| `service_overlap` | 0.10 | 与受影响服务列表的 Jaccard 相似度 |
**最大步数：** 8
### 任务 3 — Runbook 生成 `[困难]` 🔴
**代理必须做什么：** 生成一份完整的、可执行的事件 Runbook，包含所有 6 个必需的部分：
1. **诊断步骤** — 如何确认根本原因
2. **补救步骤** — 如何立即修复它
3. **回滚计划** — 如果情况恶化如何撤销修复
4. **升级标准** — 何时以及下一步通知谁
5. **预防措施** — 如何防止复发
6. **命令** — 要运行的确切 CLI 命令
**为什么它很现实：** 一份好的 Runbook 是 5 分钟修复和 2 小时停机之间的区别。它需要深厚的技术知识、操作经验，以及同时从多个角度思考问题的能力。
**奖励细分：**
| 组件 | 权重 | 描述 |
|-----------|--------|-------------|
| `runbook_diagnosis` | 0.20 | 步骤质量 × 证据相关性 |
| `runbook_remediation` | 0.25 | 步骤质量 × 命令关键词覆盖率 |
| `runbook_rollback` | 0.15 | 有回滚 + 使用回滚关键词 |
| `runbook_escalation` | 0.15 | 有标准 + 使用升级关键词 |
| `runbook_prevention` | 0.15 | 有措施 + 使用预防关键词 |
| `commands` | 0.10 | 存在真实的 CLI 命令 |
| `penalty_incomplete` | −0.05 | 缺少回滚计划 |
**最大步数：** 12
## 📐 动作空间
```
class Action(BaseModel):
action_type: ActionType # REQUIRED — see enum below
# Task 1: Alert Classification
severity: Optional[SeverityLevel] # P1 / P2 / P3 / P4
category: Optional[AlertCategory] # infrastructure/application/database/network/security
team: Optional[str] # on-call team name
# Task 2: Root Cause Analysis
root_cause_component: Optional[str] # e.g. "scheduler-service"
root_cause_type: Optional[str] # e.g. "connection_pool_exhaustion"
evidence: Optional[List[str]] # supporting log/metric observations
impact: Optional[str] # user-facing impact description
affected_services: Optional[List[str]] # list of affected service names
# Task 3: Runbook Generation
runbook: Optional[RunbookSection] # see RunbookSection schema
# Universal
notes: Optional[str] # agent reasoning / free text
confidence: Optional[float] # 0.0–1.0 self-assessed confidence
class ActionType(str, Enum):
CLASSIFY_ALERT = "classify_alert" # Terminal for Task 1
ANALYZE_LOGS = "analyze_logs" # Non-terminal: get more context
IDENTIFY_ROOT_CAUSE = "identify_root_cause" # Terminal for Task 2
GENERATE_RUNBOOK = "generate_runbook" # Terminal for Task 3
ESCALATE = "escalate" # Non-terminal: partial reward
RESOLVE = "resolve" # Terminal: marks resolution
```
## 🔭 观察空间
```
class Observation(BaseModel):
task_id: str # Episode UUID
task_type: TaskType # alert_classification / rca / runbook_generation
step: int # Current step number
max_steps: int # Episode budget
alert: Alert # The incident alert
context: IncidentContext # Service graph, recent deploys, hints
available_actions: List[str] # Valid action descriptions
progress_hint: str # Human-readable hint
episode_history: List[dict] # Last 5 steps taken
class Alert(BaseModel):
alert_id: str
title: str
description: str
service: str
environment: str # "production"
timestamp: str # ISO 8601
metrics: Dict[str, float] # key metric values
metric_details: List[MetricSnapshot]
logs: List[str] # log lines
raw_log_count: int
firing_duration_seconds: int
previous_incidents: List[str]
class IncidentContext(BaseModel):
service_graph: List[ServiceDependency] # dependency health
recent_deployments: List[dict] # what changed recently
runbook_hints: List[str] # domain knowledge hints
similar_past_incidents: List[str] # historical context
on_call_teams: Dict[str, List[str]] # team → contact list
```
## 🎯 奖励信号设计
奖励函数旨在：
1. **提供部分进度信号** — 代理可以在分析日志 (+0.05) 和升级 (+0.10) 中获得奖励，甚至在最终操作之前，从而避免稀疏奖励问题。
2. **不对称地惩罚关键失败** — 将 P1 误分类为 P4 会招致 −0.20 的惩罚（最危险的现实世界错误）。将 P4 过度分类为 P1 会招致 −0.10 的惩罚（警报疲劳）。
3. **奖励具体性** — 包含真实 CLI 命令、服务名称和技术术语的证据和 Runbook 部分比模糊的文本得分更高。
4. **施加效率压力** — 超过允许步数的 80% 会受到轻微惩罚，鼓励果断的代理。
5. **持续评分 0.0–1.0** — 绝不是二元的；始终提供关于部分进度的信息。
## 📊 场景库
该环境包含 4 个真实的生产事件场景：
| ID | 场景 | 严重性 | 类别 | 难度 |
|----|----------|----------|----------|------------|
| `db_connpool_001` | 批量作业耗尽 PostgreSQL 连接池 | P1 | 数据库 | 中等 |
| `oom_memory_002` | 无界 feature cache 导致 OOMKill 循环 | P2 | 应用程序 | 中等 |
| `cascading_timeout_003` | 网络交换机配置错误导致级联超时 | P1 | 网络 | 困难 |
| `cert_expiry_004` | TLS 证书续期被 WAF 规则阻止 | P2 | 安全 | 简单 |
每个场景包括：10-20 行真实日志、8-12 个指标、服务依赖图、最近的部署历史和完整的真实 Runbook。
## 📈 基线分数
使用 `gpt-4o-mini` 在 temperature=0.2、seed=42 下评估：
| 任务 | 难度 | 基线分数 | 备注 |
|------|-----------|---------------|-------|
| 警报分类 | 简单 | **0.71** | 在团队分配上表现挣扎 |
| 根本原因分析 | 中等 | **0.58** | 组件识别良好，证据识别薄弱 |
| Runbook 生成 | 困难 | **0.44** | 缺少回滚 + 预防部分 |
| **总体** | — | **0.58** | 仍有很大改进空间 |
## 🚀 快速开始
### 本地
```
git clone https://github.com/openenv/incident-response-env
cd incident-response-env
pip install -r requirements.txt
# 运行测试
python tests/test_env.py
# 启动 API server
uvicorn app:app --host 0.0.0.0 --port 7860
# 运行 baseline evaluation
export OPENAI_API_KEY=sk-...
python baseline/run_baseline.py
```
### Docker
```
docker build -t incident-response-env .
docker run -p 7860:7860 -e OPENAI_API_KEY=sk-... incident-response-env
# 使用自定义 seed 和 model：
docker run -p 7860:7860 -e OPENAI_API_KEY=sk-... incident-response-env \
python baseline/run_baseline.py --model gpt-4o --seed 42 --runs 3
```
### Python SDK
```
from environment.env import make_env
from environment.models import Action, ActionType, SeverityLevel, AlertCategory
# 任务 1：Alert Classification
env = make_env("alert_classification", seed=42)
obs = env.reset()
print(obs.to_agent_prompt())
action = Action(
action_type=ActionType.CLASSIFY_ALERT,
severity=SeverityLevel.P1,
category=AlertCategory.DATABASE,
team="database-team",
notes="100% connection pool utilisation, scheduler job opened 120 parallel connections",
)
obs, reward, done, info = env.step(action)
# reward.score = 0.92，reward.breakdown 显示每个 component 的分数
# 任务 3：Runbook Generation
from environment.models import RunbookSection
env3 = make_env("runbook_generation", seed=42)
obs3 = env3.reset()
action3 = Action(
action_type=ActionType.GENERATE_RUNBOOK,
runbook=RunbookSection(
diagnosis_steps=["kubectl logs -l app=scheduler-service | grep bulk-report", ...],
remediation_steps=["kubectl delete job bulk-report-export", ...],
rollback_plan=["kubectl rollout undo deployment/scheduler-service"],
escalation_criteria=["Escalate if error rate > 50% after 10 min"],
prevention_measures=["Add max_parallelism=10 to bulk jobs"],
commands=["kubectl delete job bulk-report-export"],
)
)
obs3, reward3, done3, info3 = env3.step(action3)
print(reward3.score) # 0.73
```
### HTTP API (已部署)
```
# 开始 episode
curl -X POST https://openenv-incident-response-env.hf.space/reset \
-H "Content-Type: application/json" \
-d '{"task_type": "alert_classification", "seed": 42}'
# 提交 action
curl -X POST https://openenv-incident-response-env.hf.space/step \
-H "Content-Type: application/json" \
-d '{
"session_id": "",
"action": {
"action_type": "classify_alert",
"severity": "P1",
"category": "database",
"team": "database-team",
"notes": "Connection pool exhausted"
}
}'
# 获取 episode state
curl https://openenv-incident-response-env.hf.space/state/
```
## 🧪 测试
```
python tests/test_env.py -v
# Ran 15 tests in 0.006s — OK
```
测试覆盖率：
- Reset/step/state API 正确性
- 正确和错误的分类评分
- 严重性误分类惩罚
- RCA 字段验证
- Runbook 完整性评分
- 回合终止和错误处理
- 跨 seed 的可重复性
## 🏗️ 架构决策
**为什么选择 FastAPI？** 零开销 Pydantic 集成、自动生成的 `/docs`、异步支持以及 HF Spaces 兼容性。
**为什么选择合成数据？** 真实事件数据是机密的。合成场景让我们能够精确控制真实值、确定性地改变难度，并在保持现实的同时避免隐私/法律问题。
**为什么选择基于关键词的评分？** LLM-as-judge 成本高、非确定性，并引入循环评估。关键词重叠 + 结构化评分提供了确定性、快速、可解释的评分，同时衡量了重要的内容：代理是否识别了正确的组件、正确的故障类型、正确的补救命令？
**为什么选择部分奖励？** 稀疏奖励（仅对最终操作评分）是 RL 对齐的反模式。每一步都应给代理提供有用的信号。`analyze_logs` → 小奖励。`escalate` → 小奖励。接近正确的分类 → 部分奖励。这使得环境可训练，而不仅仅是可评估。
## 📄 许可证
MIT — 见 `LICENSE`。
*为 Meta OpenEnv 黑客松构建。IncidentResponseEnv 模拟了让工程师在凌晨 3 点醒来并给公司造成数百万停机损失的真实任务。如果 AI 能擅长这一点，那就太重要了。*
标签：AIOps, AV绕过, Docker, FastAPI, HuggingFace Spaces, OpenEnv, Python, Runbook生成, SRE, 偏差过滤, 告警分诊, 安全防御评估, 强化学习环境, 故障响应, 无后门, 根因分析, 生产环境模拟, 监控指标, 站点可靠性工程, 请求拦截, 运维自动化, 逆向工具