Agent-ELK日志分析场景

Cactus li2025年10月13日...大约 6 分钟

一、介绍

在实际应用中，Agent 可结合 ELK 日志系统，基于用户诉求自主完成分析、规划、执行与结果输出，从而提升日志检索效率。

ELK（或自研方案）是互联网公司常见的分布式日志平台，用于研发在遇到线上报警或事故反馈时快速定位问题。但传统检索方式耗时较长，因此引入 Agent 辅助提效非常必要。

二、功能流程

如图，Agent-ELK 的设计使用流程图；

首先，部署一套 ELK 之后通过脚本把日志数据写入到 ELK。
然后，为该场景添加新的 AI Agent 描述话术。在执行 ELK 日志分析时，用户首先手动选择要使用的 AI Agent 服务。此时，AI Agent 会根据对应的 ELK Prompt 话术进行分析。

三、工程实现

1. 话术配置

本节更新了最新的 SQL 语句，配置了适用于 ELK 日志检索的新 Prompt，并且在 ai_client_tool_mcp 中配置了相关的 ES MCP 服务。
你可以从本节对应的分支 docs/dev-ops/mysql 获取 SQL 语句并进行更新。

2. ELK 配置

2.1 启动 ELK 服务

脚本已放在本节分支的工程目录下。你可以在本地、云服务器等具备 Docker 环境的服务上，启动 ELK 分布式日志系统。

2.2 模拟写入日志

这里为了测试，避免大家还得去配置业务系统，所以做了模拟脚本和单测代码，可以测试使用。

25-09-24.16:29:24.930 [main            ] INFO  ElkBlacklistDataTest   - 开始向Elasticsearch模拟写入拼团项目黑名单限流数据...
25-09-24.16:29:26.041 [main            ] INFO  ElkBlacklistDataTest   - 目标索引: test-buy-log-2025.09.24
25-09-24.16:29:29.731 [main            ] INFO  ElkBlacklistDataTest   - ES地址: http://192.168.1.23:9200
25-09-24.16:29:53.385 [main            ] INFO  ElkBlacklistDataTest   - Elasticsearch连接正常
25-09-24.16:29:59.591 [main            ] INFO  ElkBlacklistDataTest   - 开始生成并写入模拟数据...
25-09-24.16:30:34.685 [main            ] INFO  ElkBlacklistDataTest   - 写入第 50 条数据...
25-09-24.16:30:34.697 [main            ] INFO  ElkBlacklistDataTest   - 第 50 条数据写入成功
25-09-24.16:30:34.840 [main            ] INFO  ElkBlacklistDataTest   - 数据写入完成！成功: 50/50
25-09-24.16:30:34.847 [main            ] INFO  ElkBlacklistDataTest   - 索引信息: {"count":48,"_shards":{"total":1,"successful":1,"skipped":0,"failed":0}}
25-09-24.16:30:34.847 [main            ] INFO  ElkBlacklistDataTest   - 可以使用以下命令查看写入的数据:
25-09-24.16:30:34.847 [main            ] INFO  ElkBlacklistDataTest   - curl -X GET "http://192.168.1.23:9200/test-buy-log-2025.09.24/_search?pretty&size=5"
25-09-24.16:30:34.849 [main            ] INFO  ElkBlacklistDataTest   - 或者在Kibana中查看索引: test-buy-log-2025.09.24

执行脚本 ./elk-backlist-data.sh 会自动导入数据，也可以通过运行 ElkBlacklistDataTest 来写入数据。
这些测试数据用于 Agent-ELK 分析。你还可以根据自己的项目需求，创建不同的案例并调整 Agent Prompt 话术。

3. MCP 验证

测试案例：cn.cactusli.ai.test.spring.ai.FlowAgentMCPTest

@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest
public class FlowAgentMCPTest {

    @Test
    public void test() {
        OpenAiChatModel chatModel = OpenAiChatModel.builder()
                .openAiApi(OpenAiApi.builder()
                        .baseUrl("https://chat.199228.xyz/")
                        .apiKey("sk-xxx")
                        .completionsPath("v1/chat/completions")
                        .embeddingsPath("v1/embeddings")
                        .build())
                .defaultOptions(OpenAiChatOptions.builder()
                        .model("gpt-4o")
                        .toolCallbacks(new SyncMcpToolCallbackProvider(stdioMcpClientElasticsearch()).getToolCallbacks())
                        .build())
                .build();

        ChatResponse call = chatModel.call(Prompt.builder().messages(new UserMessage("有哪些工具可以使用")).build());
        log.info("测试结果:{}", JSON.toJSONString(call.getResult()));
    }

    /**
     * https://github.com/awesimon/elasticsearch-mcp
     * https://www.npmjs.com/package/@awesome-ai/elasticsearch-mcp
     * npm i @awesome-ai/elasticsearch-mcp
     */
    public McpSyncClient stdioMcpClientElasticsearch() {
        Map<String, String> env = new HashMap<>();
        env.put("ES_HOST", "http://192.168.1.23:9200");
        env.put("ES_API_KEY", "none");

        var stdioParams = ServerParameters.builder("npx.cmd")
                .args("-y", "@awesome-ai/elasticsearch-mcp")
                .env(env)
                .build();

        var mcpClient = McpClient.sync(new StdioClientTransport(stdioParams))
                .requestTimeout(Duration.ofSeconds(100)).build();

        var init = mcpClient.initialize();

        System.out.println("Stdio MCP Initialized: " + init);

        return mcpClient;

    }

}

{
   "metadata": {
      "contentFilters": [],
      "empty": true,
      "finishReason": "STOP"
   },
   "output": {
      "media": [],
      "messageType": "ASSISTANT",
      "metadata": {
         "role": "ASSISTANT",
         "messageType": "ASSISTANT",
         "refusal": "",
         "finishReason": "STOP",
         "annotations": [],
         "index": 0,
         "id": "chatcmpl-CJFspGiPzyUycBAlBr1gOcLrSZlpB"
      },
      "text": "以下是当前可用的工具列表，它们操作Elasticsearch相关任务：\n\n1. **列出Elasticsearch索引**：\n   - 工具名称：`functions.JavaSDKMCPClient_list_indices`\n   - 描述：列出所有可用的Elasticsearch索引，可以通过正则表达式过滤索引名称。\n\n2. **获取Elasticsearch索引的字段映射**：\n   - 工具名称：`functions.JavaSDKMCPClient_get_mappings`\n   - 描述：获取指定Elasticsearch索引的字段映射信息。\n\n3. **执行Elasticsearch搜索**：\n   - 工具名称：`functions.JavaSDKMCPClient_search`\n   - 描述：通过查询DSL在指定的Elasticsearch索引中执行搜索。\n\n4. **检查Elasticsearch集群的健康状态**：\n   - 工具名称：`functions.JavaSDKMCPClient_elasticsearch_health`\n   - 描述：获取Elasticsearch集群的健康状态，可选择是否包括索引级别的详细信息。\n\n5. **创建Elasticsearch索引**：\n   - 工具名称：`functions.JavaSDKMCPClient_create_index`\n   - 描述：在Elasticsearch中创建一个新索引，可以选择配置设置和映射。\n\n6. **创建或更新Elasticsearch索引的映射结构**：\n   - 工具名称：`functions.JavaSDKMCPClient_create_mapping`\n   - 描述：用于创建或者更新指定索引的映射（Mapping）结构。\n\n7. **批量添加数据到Elasticsearch索引**：\n   - 工具名称：`functions.JavaSDKMCPClient_bulk`\n   - 描述：将批量数据导入到指定的Elasticsearch索引中。\n\n8. **从一个索引重新索引数据到另一个索引**：\n   - 工具名称：`functions.JavaSDKMCPClient_reindex`\n   - 描述：从源索引将数据重新索引到目标索引。\n\n9. **创建或更新Elasticsearch索引模板**：\n   - 工具名称：`functions.JavaSDKMCPClient_create_index_template`\n   - 描述：用于创建或更新指定的索引模板。\n\n10. **获取Elasticsearch索引模板信息**：\n    - 工具名称：`functions.JavaSDKMCPClient_get_index_template`\n    - 描述：获取Elasticsearch中索引模板信息（可以指定模板名称过滤）。\n\n11. **删除Elasticsearch索引模板**：\n    - 工具名称：`functions.JavaSDKMCPClient_delete_index_template`\n    - 描述：删除指定的Elasticsearch索引模板。\n\n12. **并行调用多个工具**：\n    - 工具名称：`multi_tool_use.parallel`\n    - 描述：允许并行调用上面提到的多个工具，适用于可以同时运行的任务。\n\n你可以告知我你的需求，我将帮助你选择和使用合适的工具！",
      "toolCalls": []
   }
}