CrewAI 实战手记：我给 AI 排了个"值班班组"，故障诊断和写周报都不用我操心

from crewai_tools import tool
import requests
import json

@tool("错误日志检索工具")
def search_error_logs(keyword: str, minutes: int = 15) -> str:
    """根据关键词检索最近 N 分钟内的错误日志。
    输入参数：keyword 是检索关键词（如接口路径、异常类名），minutes 是时间范围（分钟）。
    返回格式化的日志摘要列表，包含时间戳、服务名、错误信息。"""
    # 实际生产中对接 ELK / Loki，这里是简化示例
    resp = requests.get(
        "http://loki.internal:3100/loki/api/v1/query_range",
        params={
            "query": f'{{job="app-logs"}} |= "{keyword}" |= "ERROR"',
            "limit": 50,
            "since": minutes * 60,
        },
        timeout=10
    )
    logs = resp.json().get("data", {}).get("result", [])

    summaries = []
    for stream in logs:
        service = stream["stream"].get("service", "unknown")
        for ts, line in stream.get("values", [])[-20:]:
            summaries.append(f"[{service}] {line[:200]}")

    if not summaries:
        return f"最近 {minutes} 分钟内未检索到包含 '{keyword}' 的错误日志"

    return f"共找到 {len(summaries)} 条相关错误日志：\n" + "\n".join(summaries[:30])


@tool("监控指标查询工具")
def query_metrics(service: str, metric: str, minutes: int = 30) -> str:
    """查询指定服务近 N 分钟的监控指标。
    参数：service 是服务名，metric 支持 qps/latency_p99/error_rate/cpu/memory。
    返回指标的当前值、峰值和趋势描述。"""
    # PromQL 查询映射
    promql_map = {
        "qps": f'sum(rate(http_requests_total{{service="{service}"}}[5m]))',
        "latency_p99": f'histogram_quantile(0.99, sum(rate(http_request_duration_seconds_bucket{{service="{service}"}}[5m])) by (le))',
        "error_rate": f'sum(rate(http_requests_total{{service="{service}",code=~"5.."}}[5m])) / sum(rate(http_requests_total{{service="{service}"}}[5m]))',
        "cpu": f'avg(cpu_usage_percent{{service="{service}"}})',
        "memory": f'avg(memory_usage_percent{{service="{service}"}})',
    }

    if metric not in promql_map:
        return f"不支持的指标类型：{metric}，可选：{list(promql_map.keys())}"

    resp = requests.get(
        "http://prometheus.internal:9090/api/v1/query",
        params={"query": promql_map[metric]},
        timeout=10
    )
    data = resp.json().get("data", {}).get("result", [])

    if not data:
        return f"未查询到 {service} 的 {metric} 数据"

    value = float(data[0]["value"][1])
    return (f"服务 {service} 的 {metric}：当前值 {value:.4f}。"
            f"（生产环境建议再补充同比/环比数据辅助判断）")

from crewai import Agent, Task, Crew, Process, LLM

# 用内网部署的开源模型，数据不出内网
local_llm = LLM(
    model="openai/qwen2.5-32b-instruct",
    base_url="http://gpu-server.internal:8000/v1",
    api_key="dummy-key",  # vLLM 不校验，随便填
    temperature=0.2,
)

# ===== 角色 1：日志分析员 =====
log_analyst = Agent(
    role="资深日志分析工程师",
    goal="从错误日志中快速定位故障的第一现场，找出最关键的报错模式和首次出现时间",
    backstory=(
        "你有 10 年的后端排障经验，特别擅长从海量日志里发现规律。"
        "你坚持一个原则：结论必须引用具体的日志证据，绝不凭感觉猜。"
        "你分析日志时总是先看报错最集中的时间窗口，再找异常堆栈的关键字。"
    ),
    tools=[search_error_logs],
    llm=local_llm,
    verbose=True,
    max_iter=5,  # 关键！限制工具调用次数，防止死循环
)

# ===== 角色 2：监控分析员 =====
metrics_analyst = Agent(
    role="SRE 监控数据分析专家",
    goal="结合日志分析员的结论，从监控指标中验证故障的影响范围和可能的根因方向",
    backstory=(
        "你是 SRE 团队的老兵，对各类监控指标的含义了如指掌。"
        "你的习惯是：拿到日志侧的线索后，一定要用指标数据交叉验证，"
        "比如日志说数据库连接超时，你就会去看连接池使用率和 DB 的 CPU。"
        "你只陈述数据支持的事实，数据不支持的说法你会明确标注'待验证'。"
    ),
    tools=[query_metrics],
    llm=local_llm,
    verbose=True,
    max_iter=5,
)

# ===== 角色 3：诊断报告撰写员 =====
report_writer = Agent(
    role="故障诊断报告撰写员",
    goal="把日志分析和监控验证的结论整合成一份值班领导 30 秒能看懂的诊断报告",
    backstory=(
        "你原来是个技术文档工程师，后来转做故障复盘。"
        "你写报告有三大铁律：先说结论再说过程、每条结论标注置信度、"
        "给出明确的下一步动作建议。你不写废话，不用'可能大概也许'糊弄人，"
        "不确定的地方就直说'信息不足，建议补充排查'。"
    ),
    llm=local_llm,
    verbose=True,
)

# ===== 任务编排 =====
task1 = Task(
    description=(
        "线上突发 5xx 告警，告警服务是 {service_name}，"
        "疑似相关接口是 {api_path}。"
        "请用错误日志检索工具，分别用接口路径和常见异常关键字检索"
        "最近 15 分钟的错误日志，总结出：1) 报错最集中的时间点；"
        "2) 最主要的异常类型和堆栈关键行；3) 涉及哪些下游服务。"
    ),
    expected_output=(
        "一段结构化的日志分析结论，包含：告警时间线、Top 异常类型（带出现次数）、"
        "涉及的下游服务清单、以及 3 条最关键的原始日志摘录。"
    ),
    agent=log_analyst,
)

task2 = Task(
    description=(
        "基于上一步日志分析员的结论：{log_findings}，"
        "用监控指标查询工具验证以下假设："
        "1) {service_name} 的错误率和 P99 延迟走势；"
        "2) 日志中涉及的下游服务的 CPU 和内存是否有异常；"
        "3) 判断故障是应用层问题还是资源层问题。"
    ),
    expected_output=(
        "监控验证结论：各指标的当前值和趋势、每个假设的验证结果"
        "（证实/证伪/数据不足）、以及初步的根因方向判断。"
    ),
    agent=metrics_analyst,
    context=[task1],  # 关键：task2 能看到 task1 的输出
)

task3 = Task(
    description=(
        "整合日志分析结论和监控验证结论，输出一份《故障初步诊断报告》。"
        "报告结构：一、结论摘要（3 句话以内）；二、故障时间线；"
        "三、根因分析（标注置信度：高/中/低）；四、影响范围；"
        "五、建议的处置动作（按优先级排序）。"
    ),
    expected_output=(
        "一份 Markdown 格式的故障诊断报告，领导读完能在 30 秒内知道"
        "发生了什么、影响多大、该派谁去处理。"
    ),
    agent=report_writer,
    context=[task1, task2],
)

# ===== 组建班组并开工 =====
diagnosis_crew = Crew(
    agents=[log_analyst, metrics_analyst, report_writer],
    tasks=[task1, task2, task3],
    process=Process.sequential,
    verbose=True,
    memory=True,  # 开启记忆，重复故障能参考历史结论
)

# 告警触发入口
def on_alert_fired(alert):
    result = diagnosis_crew.kickoff(inputs={
        "service_name": alert["service"],
        "api_path": alert["api_path"],
        "log_findings": "",  # 占位，task1 的输出会自动传给 task2
    })
    print(result)

# 模拟一次告警
on_alert_fired({
    "service": "order-service",
    "api_path": "/api/order/create",
})

from crewai import Agent, Task, Crew, Process, LLM

llm = LLM(
    model="openai/qwen2.5-32b-instruct",
    base_url="http://gpu-server.internal:8000/v1",
    api_key="dummy-key",
    temperature=0.3,
)

# ===== 三个"专员" =====
alert_specialist = Agent(
    role="告警数据专员",
    goal="从告警系统中提取本周所有告警，按严重程度和服务维度归类统计",
    backstory="你管理公司告警平台三年，对告警数据的口径烂熟于心。你统计的数据从不出错，因为你总是先核对时间范围再统计。",
    tools=[query_alerts_tool],   # 省略：查询告警平台的工具
    llm=llm,
)

change_specialist = Agent(
    role="变更记录专员",
    goal="整理本周所有上线变更，标注每次变更和后续告警的关联性",
    backstory="你在发布系统团队干了四年，深知'十次故障九次变更'。你最大的本事是把变更时间和告警时间做关联分析。",
    tools=[query_changes_tool],  # 省略：查询发布系统的工具
    llm=llm,
)

capacity_specialist = Agent(
    role="容量规划专员",
    goal="评估核心服务的资源使用趋势，识别未来两周内可能触顶的资源",
    backstory="你是做容量规划出身的，习惯用增长趋势外推而不是拍脑袋。你的报告里总有数据支撑的预警。",
    tools=[query_metrics],
    llm=llm,
)

editor = Agent(
    role="运维周报主编",
    goal="审核各专员的素材，去重合并，输出一份老板视角的运维周报",
    backstory=(
        "你是运维总监的笔杆子。老板只关心三件事：这周稳不稳、"
        "有什么风险、需要我拍板什么。你写的周报永远围绕这三件事组织，"
        "数据只保留关键的，过程细节一律砍掉。"
    ),
    llm=llm,
)

# 注意：层级模式下任务不用指定 agent，主编会自己分配
tasks = [
    Task(
        description="统计本周（{week_range}）的告警数据，按服务和严重程度归类",
        expected_output="告警统计表：各服务的告警次数、严重分布、Top 3 告警详情",
    ),
    Task(
        description="整理本周的变更记录，分析变更与告警的关联",
        expected_output="变更清单及关联分析：哪些告警疑似由变更引发",
    ),
    Task(
        description="评估核心服务容量趋势，输出风险预警",
        expected_output="容量评估结论：未来两周有触顶风险的资源清单",
    ),
    Task(
        description="汇总所有素材，撰写《运维周报》",
        expected_output="Markdown 周报：本周稳定性总结、风险与预警、需要决策的事项",
    ),
]

weekly_crew = Crew(
    agents=[alert_specialist, change_specialist, capacity_specialist, editor],
    tasks=tasks,
    process=Process.hierarchical,
    manager_llm=llm,      # 层级模式必须指定"主编"用的模型
    verbose=True,
)

result = weekly_crew.kickoff(inputs={
    "week_range": "2026-09-07 至 2026-09-13"
})