运维同学基本都接触过日志系统，最熟悉的应该就是代表性的ELK日志系统，作为日志查询的利器，我想大多数人也只有在排查问题时，才会登陆日志系统去搜索错误日志，平时肯定也懒得主动去看，去分析系统有哪些潜在的风险提示。实际看上去风平浪静，实则暗流涌动，结合上篇文章介绍的n8n工作流编排工具，我想着再把ES日志接入AI，来分析下，想看看，在平时注意不到的地方，到底还有什么隐藏故障，如下是运行后的效果图：
结果还真是发现了不少问题，这些报错，平时都没有注意过，虽然有些不影响服务使用，有些是系统不在维护，没人使用所以没人报障，看到这里，你还认为自己维护的系统是一直很稳定吗？赶快将你的日志系统接入AI进行扫描分析吧，参考工作流配置如下：
开头节点是定时器，定义每小时执行1次，即每小时查询es最新1小内的日志，这个很简单，不用多说。
节点二是运行一小段代码，用来生成当天的日期格式，因为我们的日志索引都是名称+日期的，如：nginx-access_2025.07.30，需要将这个日期与第三个节点es的插件搭配使用
from datetime import datetime# 获取输入的timestampinput_data = _input.first()timestamp = input_data.json["timestamp"]# 解析timestamp并格式化date_obj = datetime.fromisoformat(timestamp.replace('Z', '+00:00'))formatted_date = date_obj.strftime("%Y.%m.%d")# 只返回格式化后的日期return [{"json": {"date": formatted_date}}]
节点三是elasticsearch插件，主要用来连接es，配置es索引，及查询条件过滤
节点四是将查询的结果整合到一个字段，但也可以选择不整合，原因是刚发现Agent节点支持批量处理输入的数据，之前没注意，如果不合并结果，Agent节点便会每条数据单独循环处理，达不到上下文分析的效果。
# n8n大模型字符串准备器# 将清理后的消息数组转换为大模型可用的字符串格式def clean_string(text): """清理字符串中的异常引号和空字符串""" if not text: return None # 转换为字符串 text = str(text) # 处理空字符串 if text.strip() == "": return None # 处理只有引号的字符串 if text.strip() in ['""', "''", '`', '``']: return None # 处理转义的双引号 text = text.replace('\"', '"') # 处理连续的双引号 text = text.replace('""', '"') # 处理开头和结尾的引号 text = text.strip('"') text = text.strip("'") text = text.strip('`') # 处理转义字符 text = text.replace('\n', 'n') text = text.replace('\t', 't') text = text.replace('\r', 'r') # 最终检查是否为空 cleaned_text = text.strip() if not cleaned_text: return None return cleaned_text# 主处理逻辑cleaned_messages = []for item in _input.all(): # 获取当前项的数据 data = item.get("json", {}) # 提取message字段 message = data.get("message", "") # 清理字符串 cleaned_message = clean_string(message) # 只添加非空的消息 if cleaned_message is not None: cleaned_messages.append(cleaned_message)# 将数组转换为字符串if cleaned_messages: # 用换行符连接所有消息，并添加分隔符 msg_string = "n---n".join(cleaned_messages)else: msg_string = "无有效日志数据"# 返回结果msg = { "msg": msg_string, # 字符串格式，适合大模型 "count": len(cleaned_messages), "cleaned": True, "format": "string_for_ai"}# 在n8n中，最后一行会自动返回return msg
节点五便是整个工作流的大脑AI Agent，这里主要说下如何配置数据批处理模式
Max Iterations（最大迭代次数），作用：设置进程或算法运行的最大次数。
Return Intermediate Steps（返回中间步骤）
Batch Processing（批处理），定义每个批次处理的样本数量
Delay Between Batches（批次间延迟），设置一个批次完成后到下一个批次开始之间的暂停时间，单位是毫秒
另外附上提示词内容
- Role: 运维资深工程师- Background: 用户需要通过分析Elasticsearch（ES）日志内容来评估系统及服务的运行状况，提前发现潜在问题并采取预防措施。ES日志包含了丰富的系统运行信息，通过对这些日志的分析，可以及时发现异常、性能瓶颈、错误趋势等问题，从而保障系统的稳定运行。- Profile: 你是一位经验丰富的运维资深工程师，精通Elasticsearch日志分析，能够从海量日志数据中提取关键信息，识别潜在问题，并提出有效的预防措施。- Skills: 日志分析、异常检测、性能评估、趋势分析、预防性维护- Goals: 1. 从ES日志中提取关键信息，包括系统运行状态、错误信息、性能指标等。 2. 分析日志内容，识别潜在问题，如异常频繁出现、性能下降、资源瓶颈等。 3. 提供详细的系统及服务运行情况报告，包括问题描述、影响范围、建议措施等。 4. 提出预防性维护建议，帮助用户提前采取措施，避免问题恶化。 5. 以markdown语法用表格形式展示关键问题和建议，需要兼容邮件html格式。 6. 只输出结果，不要添加额外的描述。 7. 务必确保分析结果的准确性和实用性。- Constrains: 分析应基于ES日志内容，重点关注系统运行的异常和性能问题，提出的建议应具有可操作性。排除特定关键词（如`test-ok13`、`test-onr2`、`mysql`相关的内容，以及`BIRD is not ready: i/o timeout`）的日志条目。- OutputFormat: 系统及服务运行情况报告，包括问题描述、影响范围、建议措施等。- Workflow: 1. 收集并解析ES日志内容，提取关键信息。 2. 分析日志数据，识别潜在问题和异常模式。 3. 评估问题的影响范围和严重程度。 4. 提出针对性的预防性维护建议。 5. 以markdown语法用表格形式展示关键问题和建议，需要兼容邮件html格式。 6. 如果输入的日志数据为空，则直接回复"无日志信息，系统运行正常"。- Examples: - 例子1：输入ES日志内容 ```json [ {"timestamp": "2024-06-01T10:00:00Z", "level": "ERROR", "message": "Failed to connect to database", "service": "webapp", "host": "host1"}, {"timestamp": "2024-06-01T10:05:00Z", "level": "WARN", "message": "High memory usage detected", "service": "backend", "host": "host2"}, {"timestamp": "2024-06-01T10:10:00Z", "level": "INFO", "message": "System started", "service": "webapp", "host": "host1"}, {"timestamp": "2024-06-01T10:15:00Z", "level": "ERROR", "message": "Failed to connect to database", "service": "webapp", "host": "host1"} ] ``` 输出结果 # 系统日志AI分析报告nn ```markdown | 主机名 | 问题描述 | 关键性日志内容 | 影响范围 | 建议措施 | |--------|------------------------------|----------------------------------|----------------|------------------------------| | host1 | 数据库连接失败 | Failed to connect to database | webapp服务 | 检查数据库服务状态和网络连接 | | host2 | 高内存使用 | High memory usage detected | backend服务 | 监控内存使用情况，优化应用 | ``` 总结性分析 ``` 本次日志分析发现2个潜在问题，建议立即采取措施以避免影响系统稳定性。 ``` - 例子2：输入ES日志内容 ```json [ {"timestamp": "2024-06-01T10:00:00Z", "level": "INFO", "message": "System started", "service": "webapp", "host": "host1"}, {"timestamp": "2024-06-01T10:05:00Z", "level": "INFO", "message": "System running normally", "service": "backend", "host": "host2"} ] ``` 输出结果 ```markdown | 主机名 | 问题描述 | 关键性日志内容 | 影响范围 | 建议措施 | ``` 总结性分析 ``` 无日志信息，系统运行正常。 ```
整个工作流内容基本结束了，这是一个很简单很基础的demo样例，看到这，你可能会有疑问，哪怕1小时es的日志量也有成千上万行，这么多日志一下子都投喂给大模型，消耗的token很大，成本有点高，而且日志里不一定都是有用的日志，这么一股脑的都扔进去有点太low了，是的，说的很对，首先，我需要说明下，上面的例子是针对系统message日志进行采样分析，而不是业务日志，message日志量本身不大，而且大部分应用组件报错都会输出到message日志，所以针对运维来说，用message日志分析应用组件是个比较好的切入口，另外，刚开始，大家应该不知道系统日志里到底都有什么，哪些该过滤，哪些该保留，所以，刚开始不妨大胆些，先投喂AI看输出结果，再进行筛选，这样虽然不知道应该保留哪些日志给AI，但起码知道，应该去掉那些日志不输入给AI。如果上面的例子，算是1.0版本，那么请看下面的2.0版本。
新增了一个节点，用来筛除es不必要日志，想屏蔽那些日志，把关键字或词语都填进去即可，然后es节点查询语句引用前面的过滤条件
"must_not": [ { "terms": { "message": {{ $json.message }} } } ]
有了2.0版本，还有3.0版本，如果环境比较多，应用组件种类也多，比如k8s相关组件日志是一组，db数据库组件日志又是一组，基础架构组件还是一组，怎么分类呢，如下
当然，实现方式不止一种，n8n丰富的插件也不止这些功能和应用场景，感兴趣的小伙伴，可以私信交流使用场景及优化建议，期待将AI应用到实际运维工作中，改变传统运维模式，解放双手，提高效率！
