开发 AI 写代码、运维 AI 排故障，你的岗位 AI 提效方案是什么？

@Entity
@Table(name = "user_tag")
@Data
public class UserTag {

    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    @Column(nullable = false, length = 64)
    private String tagName;

    @Column(length = 256)
    private String description;

    @Column(nullable = false)
    private Integer tagType; // 1-手动 2-规则

    @Column(name = "created_at", updatable = false)
    private LocalDateTime createdAt;

    @Column(name = "updated_at")
    private LocalDateTime updatedAt;

    @PrePersist
    protected void onCreate() {
        createdAt = LocalDateTime.now();
        updatedAt = createdAt;
    }

    @PreUpdate
    protected void onUpdate() {
        updatedAt = LocalDateTime.now();
    }
}

@RestController
@RequestMapping("/api/v1/user-tags")
@RequiredArgsConstructor
public class UserTagController {

    private final UserTagService userTagService;

    @PostMapping
    public Result<Long> createTag(@Valid @RequestBody TagCreateReq req) {
        return Result.success(userTagService.createTag(req));
    }

    @GetMapping("/{id}")
    public Result<TagVO> getTag(@PathVariable Long id) {
        return Result.success(userTagService.getTagById(id));
    }

    @GetMapping
    public Result<PageResult<TagVO>> listTags(TagQueryReq req) {
        return Result.success(userTagService.pageQuery(req));
    }

    @PutMapping("/{id}")
    public Result<Void> updateTag(@PathVariable Long id,
                                   @Valid @RequestBody TagUpdateReq req) {
        userTagService.updateTag(id, req);
        return Result.success();
    }

    @DeleteMapping("/{id}")
    public Result<Void> deleteTag(@PathVariable Long id) {
        userTagService.deleteTag(id);
        return Result.success();
    }

    @PostMapping("/batch-assign")
    public Result<Void> batchAssign(@Valid @RequestBody BatchAssignReq req) {
        userTagService.batchAssign(req.getUserIds(), req.getTagIds());
        return Result.success();
    }
}

@Service
@RequiredArgsConstructor
public class UserTagServiceImpl implements UserTagService {

    private final UserTagRepository userTagRepository;
    private final UserTagRelationRepository relationRepository;
    private final ApplicationEventPublisher eventPublisher;

    @Override
    @Transactional
    public Long createTag(TagCreateReq req) {
        // 同名校验
        if (userTagRepository.existsByTagName(req.getTagName())) {
            throw new BizException("TAG_NAME_DUPLICATED", "标签名已存在");
        }
        UserTag tag = new UserTag();
        tag.setTagName(req.getTagName());
        tag.setDescription(req.getDescription());
        tag.setTagType(req.getTagType() != null ? req.getTagType() : 1);
        userTagRepository.save(tag);
        return tag.getId();
    }

    @Override
    @Transactional
    public void batchAssign(List<Long> userIds, List<Long> tagIds) {
        if (CollectionUtils.isEmpty(userIds) || CollectionUtils.isEmpty(tagIds)) {
            throw new BizException("PARAM_INVALID", "用户ID或标签ID不能为空");
        }
        // 批量插入，去重，避免重复打标
        List<UserTagRelation> relations = new ArrayList<>();
        for (Long userId : userIds) {
            for (Long tagId : tagIds) {
                relations.add(new UserTagRelation(userId, tagId));
            }
        }
        relationRepository.batchInsertIgnoreDuplicate(relations);

        // 发事件通知下游（比如刷新用户画像缓存）
        eventPublisher.publishEvent(new TagAssignedEvent(userIds, tagIds));
    }
}

@SpringBootTest
class UserTagServiceImplTest {

    @Autowired
    private UserTagService userTagService;

    @MockBean
    private UserTagRepository userTagRepository;

    @Test
    void createTag_shouldSuccess_whenNameNotDuplicated() {
        TagCreateReq req = new TagCreateReq();
        req.setTagName("高价值用户");
        req.setDescription("月消费大于5000");

        when(userTagRepository.existsByTagName("高价值用户")).thenReturn(false);
        when(userTagRepository.save(any())).thenAnswer(inv -> {
            UserTag t = inv.getArgument(0);
            t.setId(1L);
            return t;
        });

        Long id = userTagService.createTag(req);
        assertNotNull(id);
        assertEquals(1L, id);
    }

    @Test
    void createTag_shouldThrow_whenNameDuplicated() {
        TagCreateReq req = new TagCreateReq();
        req.setTagName("VIP");

        when(userTagRepository.existsByTagName("VIP")).thenReturn(true);

        BizException ex = assertThrows(BizException.class,
                () -> userTagService.createTag(req));
        assertEquals("TAG_NAME_DUPLICATED", ex.getCode());
    }

    @Test
    void batchAssign_shouldThrow_whenEmptyInput() {
        assertThrows(BizException.class,
                () -> userTagService.batchAssign(Collections.emptyList(), List.of(1L)));
        assertThrows(BizException.class,
                () -> userTagService.batchAssign(List.of(1L), Collections.emptyList()));
    }
}

import os
import json
from datetime import datetime, timedelta
from openai import OpenAI

client = OpenAI(api_key=os.getenv("LLM_API_KEY"))

def collect_diagnostic_context(service_name, alert_time, window_minutes=15):
    """
    汇聚故障现场的多源信息：日志、监控指标、最近发布记录、依赖服务状态
    """
    start = alert_time - timedelta(minutes=window_minutes)
    end = alert_time + timedelta(minutes=5)

    context = {
        "service": service_name,
        "alert_time": alert_time.isoformat(),
        "window": f"{start.isoformat()} ~ {end.isoformat()}",
        "error_logs": fetch_error_logs(service_name, start, end, limit=50),
        "metrics": fetch_metrics_snapshot(service_name, start, end),
        "recent_deploys": fetch_recent_deploys(service_name, hours=2),
        "downstream_status": check_downstream(service_name),
    }
    return context


def fetch_error_logs(service, start, end, limit=50):
    # 实际项目中对接 ELK / Loki / 云日志服务
    # 这里给出简化示例
    return [
        "[ERROR] 2026-08-05 02:03:11 order-service DB connection timeout",
        "[WARN]  2026-08-05 02:03:12 order-service fallback to default stock",
        "[ERROR] 2026-08-05 02:03:15 order-service RPC to inventory-service timeout",
    ]


def fetch_metrics_snapshot(service, start, end):
    return {
        "p99_latency_ms": {"baseline": 200, "current": 3000},
        "qps": {"baseline": 1200, "current": 800},
        "cpu_usage": {"baseline": 45, "current": 72},
        "db_connection_pool": {"max": 50, "active": 50, "waiting": 23},
    }


def fetch_recent_deploys(service, hours=2):
    return [
        {"version": "v2.3.1", "time": "2026-08-05T01:50:00", "changes": "库存查询改RPC"},
    ]


def check_downstream(service):
    return {
        "inventory-service": {"status": "degraded", "latency_p99_ms": 2800},
        "user-service": {"status": "healthy", "latency_p99_ms": 80},
        "payment-service": {"status": "healthy", "latency_p99_ms": 150},
    }


def ai_diagnose(context):
    """
    把聚合后的现场信息喂给大模型，让它给出初步判断和处置建议
    """
    prompt = f"""你是一位资深 SRE 工程师。以下是某次线上故障的现场信息，请给出：
1. 最可能的根因判断（按可能性排序，最多3条）
2. 每条根因的依据
3. 建议的处置动作（分紧急止血 / 根因修复两步）
4. 需要人工进一步确认的点

现场信息（JSON）：
{json.dumps(context, ensure_ascii=False, indent=2)}
"""
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.2,
    )
    return resp.choices[0].message.content


if __name__ == "__main__":
    alert_time = datetime(2026, 8, 5, 2, 3, 0)
    ctx = collect_diagnostic_context("order-service", alert_time)
    report = ai_diagnose(ctx)
    print(report)

import smtplib
from email.mime.text import MIMEText

def daily_health_check():
    items = []

    # 磁盘使用率
    disk = check_disk_usage()
    if disk["usage_percent"] > 80:
        items.append(f"⚠️ 磁盘使用率 {disk['usage_percent']}%，超过80%阈值")

    # 证书有效期
    certs = check_cert_expiry()
    for c in certs:
        if c["days_left"] < 14:
            items.append(f"⚠️ 证书 {c['domain']} 剩余 {c['days_left']} 天")

    # 备份完成情况
    backup = check_last_backup()
    if not backup["success"]:
        items.append(f"❌ 昨夜备份失败：{backup['reason']}")

    # 异常实例
    dead = check_dead_instances()
    if dead:
        items.append(f"❌ 以下实例不可达：{', '.join(dead)}")

    summary = "\n".join(items) if items else "✅ 今日巡检一切正常"
    push_to_dingtalk(summary)
    return summary


def push_to_dingtalk(msg):
    # 调用钉钉/企业微信机器人 webhook
    webhook = os.getenv("OPS_DINGTALK_WEBHOOK")
    # ...省略具体推送逻辑
    pass


if __name__ == "__main__":
    print(daily_health_check())

def generate_regression_cases(diff_text, requirement_desc):
    prompt = f"""你是测试工程师。下面是一段代码变更和对应的需求描述，
请生成回归测试用例，要求：
1. 覆盖正常路径、异常路径、边界值
2. 标注每条用例的优先级（P0/P1/P2）
3. 给出预期结果

需求描述：{requirement_desc}

代码变更：
{diff_text}
"""
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.3,
    )
    return resp.choices[0].message.content

# AI 生成的 Playwright 脚本示例
from playwright.sync_api import sync_playwright
import pytest

def test_login_flow(page):
    page.goto("https://app.example.com/login")
    page.fill("#username", "testuser")
    page.fill("#password", "Test@1234")
    page.click("button[type=submit]")

    # 断言登录成功后跳转到首页
    page.wait_for_url("**/dashboard")
    assert page.title() == "工作台"

    # 断言顶部用户名展示正确
    username_text = page.text_content(".user-name")
    assert username_text == "测试用户"

def generate_prd_draft(voice_note, template):
    prompt = f"""你是资深产品经理。以下是我口述的需求背景和场景，
请按模板整理成 PRD 草稿，包含：
1. 需求背景与目标
2. 用户故事
3. 功能清单（按优先级 P0/P1/P2）
4. 核心流程描述
5. 验收标准
6. 风险与依赖

口述内容：{voice_note}

PRD模板：{template}
"""
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.4,
    )
    return resp.choices[0].message.content

def text_to_sql(question, schema_meta):
    prompt = f"""你是数据分析师。根据以下表结构元数据，把自然语言问题翻译成 SQL。
要求：
1. 只能使用提供的表和字段
2. 加上 LIMIT 防止全表扫描
3. 给出 SQL 的简要解释

表结构元数据：
{schema_meta}

问题：{question}
"""
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.0,
    )
    return resp.choices[0].message.content

-- 问题：上个月消费超过5000元的用户有多少？其中多少是老客？
SELECT
    COUNT(*) AS total_users,
    SUM(CASE WHEN u.register_time < DATE_SUB('2026-07-01', INTERVAL 30 DAY)
             THEN 1 ELSE 0 END) AS old_users
FROM (
    SELECT user_id, SUM(amount) AS total_amount
    FROM order_table
    WHERE create_time >= '2026-07-01'
      AND create_time < '2026-08-01'
      AND status = 'PAID'
    GROUP BY user_id
    HAVING SUM(amount) > 5000
) t
JOIN user_table u ON u.id = t.user_id;