用 AI 做智能日志分析——自动从海量日志中提取错误根因和修复建议

logai/
├── logai.py             # CLI 主程序
├── analyzer.py           # AI 日志分析
└── requirements.txt

# analyzer.py
import re
from collections import Counter
from openai import OpenAI
import os
from dotenv import load_dotenv
load_dotenv()

client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
)

def extract_errors(log_content: str) -> dict:
    """从日志中提取错误信息。"""
    lines = log_content.split("\n")

    errors = []
    error_patterns = [
        (r"(ERROR|FATAL|CRITICAL|Exception|Traceback).*", "错误"),
        (r"\d{3}\s+(Internal Server Error|Bad Gateway|Service Unavailable)", "HTTP错误"),
        (r"(timeout|timed out|connection refused)", "超时/连接"),
        (r"(OutOfMemory|MemoryError|OOM)", "内存"),
        (r"(Permission denied|Access denied|Forbidden)", "权限"),
    ]

    for line in lines:
        for pattern, category in error_patterns:
            if re.search(pattern, line, re.IGNORECASE):
                errors.append({"line": line.strip()[:200], "category": category})
                break

    # 统计错误类型
    categories = Counter(e["category"] for e in errors)

    return {
        "total_lines": len(lines),
        "error_count": len(errors),
        "categories": dict(categories),
        "sample_errors": errors[:20],
    }


def analyze_root_cause(log_content: str, stats: dict) -> str:
    """AI 分析根因和建议。"""
    # 取日志的前 2000 字符 + 错误样本
    sample = log_content[:2000]
    errors = "\n".join(
        f"[{e['category']}] {e['line'][:150]}"
        for e in stats["sample_errors"][:10]
    )

    prompt = f"""你是运维专家。分析以下日志，找出根本原因和修复建议。

## 统计信息
总行数：{stats['total_lines']}
错误数：{stats['error_count']}
错误分布：{stats['categories']}

## 日志样本
{sample}

## 错误详情
{errors}

请按以下格式输出：
1. 🔴 根本原因：一句话指出最可能的问题
2. 🟡 影响范围：影响了哪些服务或功能
3. 🟢 修复建议：具体可操作的修复步骤
4. 🔵 预防措施：如何避免类似问题
"""
    resp = client.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.2,
        max_tokens=1500,
    )
    return resp.choices[0].message.content

#!/usr/bin/env python3
# logai.py
import sys
import click
from pathlib import Path
from analyzer import extract_errors, analyze_root_cause

@click.group()
def cli(): pass

@cli.command()
@click.argument("filepath", type=click.Path(exists=True))
@click.option("--errors-only", "-e", is_flag=True, help="只显示错误统计")
def analyze(filepath, errors_only):
    """分析日志文件。"""
    with open(filepath, "r", encoding="utf-8", errors="ignore") as f:
        content = f.read()

    stats = extract_errors(content)
    click.echo(f"📊 日志分析：{filepath}")
    click.echo(f"   总行数：{stats['total_lines']}")
    click.echo(f"   错误数：{stats['error_count']}")
    click.echo(f"   错误分布：{stats['categories']}")

    if errors_only:
        return

    click.echo("\n🤖 AI 分析中...")
    result = analyze_root_cause(content, stats)
    click.echo(f"\n{result}")

if __name__ == "__main__":
    cli()

# 分析日志
python logai.py analyze /var/log/app.log
# 📊 日志分析：/var/log/app.log
#    总行数：8342
#    错误数：47
#    错误分布：{'错误': 23, '超时/连接': 12, '内存': 8, '权限': 4}
#
# 1. 🔴 根本原因：数据库连接池耗尽导致请求超时
# 2. 🟡 影响范围：API 服务的用户认证和订单查询接口
# 3. 🟢 修复建议：增加连接池大小到 20，添加连接超时回收机制

# 只看统计
python logai.py analyze /var/log/app.log --errors-only

# 支持标准输入，方便管道使用
@cli.command()
@click.argument("filepath", required=False)
def tail(filepath):
    """分析最新的 500 行日志。"""
    if filepath:
        lines = Path(filepath).read_text().split("\n")[-500:]
    else:
        lines = sys.stdin.read().split("\n")[-500:]
    # ... 同上分析逻辑

tail -n 500 /var/log/app.log | python logai.py tail