1
<dependencies>
2
    <!-- LangChain4j Spring Boot Starter -->
3
    <dependency>
4
        <groupId>dev.langchain4j</groupId>
5
        <artifactId>langchain4j-spring-boot-starter</artifactId>
6
        <version>0.35.0</version>
7
    </dependency>
8

9
    <!-- OpenAI 集成 (可替换为其他模型提供商) -->
10
    <dependency>
11
        <groupId>dev.langchain4j</groupId>
12
        <artifactId>langchain4j-open-ai-spring-boot-starter</artifactId>
13
        <version>0.35.0</version>
14
    </dependency>
15
</dependencies>

1
import dev.langchain4j.service.SystemMessage;
2
import dev.langchain4j.service.UserMessage;
3
import dev.langchain4j.service.MemoryId;
4
import dev.langchain4j.service.TokenStream;
5

6
public interface DataAnalystAgent {
7

8
    @SystemMessage("""
9
        你是一个数据分析助手。你的职责是:
10
        1. 根据用户问题,选择合适的工具查询数据
11
        2. 分析数据趋势,识别异常模式
12
        3. 生成专业的分析报告
13

14
        【可用工具】
15
        - queryStatistics: 查询统计数据
16
        - analyzeTrend: 分析趋势
17
        - generateReport: 生成报告
18

19
        【工作流程】
20
        1. 理解用户问题,确定需要哪些数据
21
        2. 依次调用工具收集原始数据
22
        3. 进行分析和总结
23
        4. 用专业但易懂的语言回复用户
24
        """)
25
    TokenStream chat(
26
        @MemoryId String conversationId,
27
        @UserMessage String question
28
    );
29
}

1
import dev.langchain4j.agent.tool.Tool;
2
import dev.langchain4j.agent.tool.P;
3
import lombok.RequiredArgsConstructor;
4
import org.springframework.stereotype.Component;
5

6
@Component
7
@RequiredArgsConstructor
8
public class DataQueryTools {
9

10
    private final DataRepository dataRepository;
11

12
    @Tool("查询指定时间范围内的统计数据")
13
    public String queryStatistics(
14
        @P("开始日期,格式 YYYY-MM-DD") String startDate,
15
        @P("结束日期,格式 YYYY-MM-DD") String endDate
16
    ) {
17
        // 查询数据库
18
        List<DataRecord> records = dataRepository.findByDateBetween(startDate, endDate);
19

20
        // 统计分析
21
        long totalCount = records.size();
22
        double avgValue = records.stream()
23
            .mapToDouble(DataRecord::getValue)
24
            .average()
25
            .orElse(0.0);
26

27
        // 返回格式化结果
28
        return String.format(
29
            "时间范围: %s 至 %s\n" +
30
            "记录总数: %d 条\n" +
31
            "平均值: %.2f",
32
            startDate, endDate, totalCount, avgValue
33
        );
34
    }
35

36
    @Tool("查询最近N天的趋势数据")
37
    public String queryTrend(
38
        @P("查询天数,例如 7 表示最近7天") int days
39
    ) {
40
        LocalDate endDate = LocalDate.now();
41
        LocalDate startDate = endDate.minusDays(days);
42

43
        List<DataRecord> records = dataRepository.findByDateBetween(startDate, endDate);
44

45
        // 按日期分组统计
46
        Map<LocalDate, Double> dailyAvg = records.stream()
47
            .collect(Collectors.groupingBy(
48
                DataRecord::getDate,
49
                Collectors.averagingDouble(DataRecord::getValue)
50
            ));
51

52
        // 格式化输出
53
        StringBuilder result = new StringBuilder("每日趋势:\n");
54
        dailyAvg.entrySet().stream()
55
            .sorted(Map.Entry.comparingByKey())
56
            .forEach(entry -> {
57
                result.append(String.format(
58
                    "%s: %.2f\n",
59
                    entry.getKey(), entry.getValue()
60
                ));
61
            });
62

63
        return result.toString();
64
    }
65
}

1
@Bean
2
public DataAnalystAgent dataAnalystAgent(
3
    StreamingChatLanguageModel model,
4
    ChatHistoryService chatHistoryService
5
) {
6
    return AiServices.builder(DataAnalystAgent.class)
7
        .streamingChatLanguageModel(model)
8
        .chatMemoryProvider(memoryId -> {
9
            MessageWindowChatMemory memory = MessageWindowChatMemory.withMaxMessages(20);
10

11
            // 从数据库加载历史消息
12
            List<ChatMessage> history = chatHistoryService.loadRecentMessages(
13
                memoryId, 20
14
            );
15

16
            // 恢复到内存
17
            for (ChatMessage msg : history) {
18
                if ("user".equals(msg.getRole())) {
19
                    memory.add(UserMessage.from(msg.getContent()));
20
                } else if ("assistant".equals(msg.getRole())) {
21
                    memory.add(AiMessage.from(msg.getContent()));
22
                }
23
            }
24

25
            return memory;
26
        })
27
        .tools(dataQueryTools, analysisTools)
28
        .build();
29
}

1
import org.springframework.http.MediaType;
2
import org.springframework.http.codec.ServerSentEvent;
3
import org.springframework.web.bind.annotation.*;
4
import reactor.core.publisher.Flux;
5

6
@RestController
7
@RequestMapping("/api/v1/analyst")
8
@RequiredArgsConstructor
9
public class AnalystController {
10

11
    private final DataAnalystAgent agent;
12
    private final ChatHistoryService chatHistoryService;
13

14
    @PostMapping(value = "/chat-stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
15
    public Flux<ServerSentEvent<String>> chatStream(
16
        @RequestParam String question,
17
        @RequestParam String conversationId
18
    ) {
19
        return Flux.create(sink -> {
20
            try {
21
                TokenStream tokenStream = agent.chat(conversationId, question);
22
                StringBuilder fullResponse = new StringBuilder();
23

24
                tokenStream
25
                    .onNext(token -> {
26
                        // 实时推送每个 Token
27
                        fullResponse.append(token);
28
                        ServerSentEvent<String> event = ServerSentEvent.<String>builder()
29
                            .event("message")
30
                            .data(token)
31
                            .build();
32
                        sink.next(event);
33
                    })
34
                    .onToolExecuted(toolExecution -> {
35
                        // 推送工具调用信息
36
                        String toolInfo = String.format(
37
                            "[工具调用: %s]",
38
                            toolExecution.request().name()
39
                        );
40
                        ServerSentEvent<String> event = ServerSentEvent.<String>builder()
41
                            .event("tool")
42
                            .data(toolInfo)
43
                            .build();
44
                        sink.next(event);
45
                    })
46
                    .onComplete(response -> {
47
                        // 保存对话历史
48
                        chatHistoryService.saveMessage(
49
                            conversationId, "user", question
50
                        );
51
                        chatHistoryService.saveMessage(
52
                            conversationId, "assistant", fullResponse.toString()
53
                        );
54

55
                        // 发送完成事件
56
                        ServerSentEvent<String> event = ServerSentEvent.<String>builder()
57
                            .event("done")
58
                            .data("")
59
                            .build();
60
                        sink.next(event);
61
                        sink.complete();
62
                    })
63
                    .onError(error -> {
64
                        // 错误处理
65
                        ServerSentEvent<String> event = ServerSentEvent.<String>builder()
66
                            .event("error")
67
                            .data("服务暂时不可用")
68
                            .build();
69
                        sink.next(event);
70
                        sink.error(error);
71
                    })
72
                    .start();
73

74
            } catch (Exception e) {
75
                sink.error(e);
76
            }
77
        });
78
    }
79
}

1
@Component
2
public class ReportTools {
3
    @Tool("生成 Markdown 格式的分析报告")
4
    public String generateMarkdownReport(
5
        String title,
6
        String statistics,
7
        String trendAnalysis,
8
        String anomalies,
9
        String conclusions
10
    ) {
11
        return String.format("""
12
            # %s
13

14
            ## 📊 统计数据概览
15
            %s
16

17
            ## 📈 趋势分析
18
            %s
19

20
            ## ⚠️ 异常检测
21
            %s
22

23
            ## 💡 结论与建议
24
            %s
25
            """,
26
            title, statistics, trendAnalysis, anomalies, conclusions
27
        );
28
    }
29

30
    @Tool("生成趋势折线图(Mermaid 格式)")
31
    public String generateTrendChart(String trendData) {
32
        // 解析数据并生成 Mermaid 图表代码
33
    }
34
}

1
@SystemMessage("""
2
    你是数据分析助手。当用户要求生成报告时,请按以下流程操作:
3

4
    1. 调用 queryStatistics 查询原始数据
5
    2. 调用 analyzeTrend 分析趋势
6
    3. 调用 detectAnomalies 检测异常
7
    4. 调用 generateTrendChart 生成图表
8
    5. 调用 generateMarkdownReport 组装报告
9
    6. 将完整报告返回给用户
10

11
    重要规则:
12
    - 必须先查询数据,再进行分析
13
    - 图表数据必须来自查询结果,不要编造
14
    - 报告要包含图表可视化
15
    """)
16
TokenStream chat(@MemoryId String conversationId, @UserMessage String question);

1
@SystemMessage("""
2
    你是数据分析助手。职责:
3
    1. 查询数据: 使用 queryStatistics 工具
4
    2. 分析趋势: 使用 analyzeTrend 工具
5
    3. 生成报告: 使用 generateReport 工具
6

7
    工作流程:
8
    1. 理解用户问题
9
    2. 依次调用工具
10
    3. 综合分析结果
11
    4. 用专业语言回复
12

13
    重要规则:
14
    - 先查询数据,再分析
15
    - 数字保留2位小数
16
    - 不要编造数据
17
    """)

1
@Tool("查询统计数据")
2
public String queryStatistics(String startDate, String endDate) {
3
    try {
4
        LocalDate start = LocalDate.parse(startDate);
5
        LocalDate end = LocalDate.parse(endDate);
6

7
        if (end.isBefore(start)) {
8
            return "错误: 结束日期不能早于开始日期";
9
        }
10

11
        // 执行查询...
12

13
    } catch (DateTimeParseException e) {
14
        return "错误: 日期格式不正确,请使用 YYYY-MM-DD 格式";
15
    } catch (Exception e) {
16
        return "错误: 查询失败,请稍后重试";
17
    }
18
}

1
public interface DataAnalystAgent {
2
    @SystemMessage("""
3
        你是数据分析助手。职责:
4
        1. 查询数据 → queryStatistics
5
        2. 分析趋势 → analyzeTrend
6
        3. 检测异常 → detectAnomalies
7
        4. 生成报告 → generateReport
8

9
        工作流程:
10
        1. 理解用户问题
11
        2. 依次调用工具
12
        3. 综合分析结果
13
        4. 用专业语言回复
14
        """)
15
    TokenStream chat(@MemoryId String conversationId, @UserMessage String question);
16
}

1
@Component
2
@RequiredArgsConstructor
3
public class DataQueryTools {
4
    private final DataRepository dataRepository;
5

6
    @Tool("查询统计数据")
7
    public String queryStatistics(
8
        @P("开始日期,格式 YYYY-MM-DD") String startDate,
9
        @P("结束日期,格式 YYYY-MM-DD") String endDate
10
    ) {
11
        List<DataRecord> records = dataRepository.findByDateBetween(startDate, endDate);
12
        return formatStatistics(records);
13
    }
14
}
15

16
@Component
17
public class AnalysisTools {
18
    @Tool("分析趋势")
19
    public String analyzeTrend(@P("趋势数据") String trendData) {
20
        // 分析逻辑...
21
    }
22

23
    @Tool("检测异常")
24
    public String detectAnomalies(@P("每日数据") String dailyData) {
25
        // 异常检测逻辑...
26
    }
27
}

1
@Configuration
2
@RequiredArgsConstructor
3
public class LangChain4jConfig {
4
    private final DataQueryTools dataQueryTools;
5
    private final AnalysisTools analysisTools;
6

7
    @Bean
8
    public DataAnalystAgent dataAnalystAgent(StreamingChatLanguageModel model) {
9
        return AiServices.builder(DataAnalystAgent.class)
10
            .streamingChatLanguageModel(model)
11
            .chatMemoryProvider(memoryId ->
12
                MessageWindowChatMemory.withMaxMessages(20)
13
            )
14
            .tools(dataQueryTools, analysisTools)
15
            .build();
16
    }
17
}

1
@RestController
2
@RequestMapping("/api/v1/analyst")
3
@RequiredArgsConstructor
4
public class AnalystController {
5
    private final DataAnalystAgent agent;
6

7
    @PostMapping(value = "/chat", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
8
    public Flux<ServerSentEvent<String>> chat(
9
        @RequestParam String question,
10
        @RequestParam String conversationId
11
    ) {
12
        return Flux.create(sink -> {
13
            TokenStream stream = agent.chat(conversationId, question);
14

15
            stream
16
                .onNext(token -> sink.next(
17
                    ServerSentEvent.<String>builder()
18
                        .event("message")
19
                        .data(token)
20
                        .build()
21
                ))
22
                .onComplete(response -> {
23
                    sink.next(
24
                        ServerSentEvent.<String>builder()
25
                            .event("done")
26
                            .data("")
27
                            .build()
28
                    );
29
                    sink.complete();
30
                })
31
                .start();
32
        });
33
    }
34
}