开头节点是定时器，定义每小时执行1次，即每小时查询es最新1小内的日志，这个很简单，不用多说。
节点二是运行一小段代码，用来生成当天的日期格式，因为我们的日志索引都是名称+日期的，如：nginx-access_2025.07.30，需要将这个日期与第三个节点es的插件搭配使用
fromdatetimeimportdatetime#获取输入的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大模型字符串准备器#将清理后的消息数组转换为大模型可用的字符串格式defclean_string(text):"""清理字符串中的异常引号和空字符串"""ifnottext:returnNone#转换为字符串text=str(text)#处理空字符串iftext.strip()=="":returnNone#处理只有引号的字符串iftext.strip()in['""',"''",'`','``']:returnNone#处理转义的双引号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()ifnotcleaned_text:returnNonereturncleaned_text#主处理逻辑cleaned_messages=[]foritemin_input.all():#获取当前项的数据data=item.get("json",{})#提取message字段message=data.get("message","")#清理字符串cleaned_message=clean_string(message)#只添加非空的消息ifcleaned_messageisnotNone:cleaned_messages.append(cleaned_message)#将数组转换为字符串ifcleaned_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中，最后一行会自动返回returnmsg节点五便是整个工作流的大脑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`相关的内容，以及`BIRDisnotready:i/otimeout`）的日志条目。-OutputFormat:系统及服务运行情况报告，包括问题描述、影响范围、建议措施等。-Workflow:1.收集并解析ES日志内容，提取关键信息。2.分析日志数据，识别潜在问题和异常模式。3.评估问题的影响范围和严重程度。4.提出针对性的预防性维护建议。5.以markdown语法用表格形式展示关键问题和建议，需要兼容邮件html格式。6.如果输入的日志数据为空，则直接回复"无日志信息，系统运行正常"。-Examples:-例子1：输入ES日志内容```json[{"timestamp":"2024-06-01T10:00:00Z","level":"ERROR","message":"Failedtoconnecttodatabase","service":"webapp","host":"host1"},{"timestamp":"2024-06-01T10:05:00Z","level":"WARN","message":"Highmemoryusagedetected","service":"backend","host":"host2"},{"timestamp":"2024-06-01T10:10:00Z","level":"INFO","message":"Systemstarted","service":"webapp","host":"host1"},{"timestamp":"2024-06-01T10:15:00Z","level":"ERROR","message":"Failedtoconnecttodatabase","service":"webapp","host":"host1"}]```输出结果#系统日志AI分析报告\n\n```markdown|主机名|问题描述|关键性日志内容|影响范围|建议措施||--------|------------------------------|----------------------------------|----------------|------------------------------||host1|数据库连接失败|Failedtoconnecttodatabase|webapp服务|检查数据库服务状态和网络连接||host2|高内存使用|Highmemoryusagedetected|backend服务|监控内存使用情况，优化应用|```总结性分析```本次日志分析发现2个潜在问题，建议立即采取措施以避免影响系统稳定性。```-例子2：输入ES日志内容```json[{"timestamp":"2024-06-01T10:00:00Z","level":"INFO","message":"Systemstarted","service":"webapp","host":"host1"},{"timestamp":"2024-06-01T10:05:00Z","level":"INFO","message":"Systemrunningnormally","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应用到实际运维工作中，改变传统运维模式，解放双手，提高效率！
