Agent任务分解：大任务拆解与并行执行的Java实践
2026/5/29大约 16 分钟Agent任务分解并行执行DAGSpring AIJava
Agent任务分解：大任务拆解与并行执行的Java实践
开篇故事：120分钟 vs 8分钟的差距
2025年年底，王强（法律科技公司Java工程师，工作3年）接到一个任务：用AI分析公司积累的100份劳动合同，找出其中的风险条款。
这是个典型的AI能干好的活——让AI逐一阅读合同，检查付款条款、违约金条款、竞业限制条款是否合规。
第一版实现：串行处理
合同1 → AI分析 → 1.2分钟
合同2 → AI分析 → 1.1分钟
...
合同100 → AI分析 → 1.2分钟
总计：约120分钟两小时……用户不接受。
王强的第一反应：加并发。但问题来了：
- 100个请求同时打到AI API，瞬间触发Rate Limit
- 某几份合同特别长（150页），处理时间是普通合同的5倍
- 部分合同需要依赖前一份的分析结果（例如关联合同、补充协议）
- 某份合同分析失败，整批任务怎么处理？
这不是简单的并发问题，这是任务分解与调度问题。
王强花了2周设计了一套基于DAG的任务调度系统：
- 100份合同分成20个批次，每批5份并行
- 依赖关系用DAG表达，保证关联合同顺序执行
- 失败的合同单独重试，不影响其他批次
- 实时进度展示，用户可以看到"已完成43/100"
最终：120分钟 → 8分钟。
一、任务分解的4种核心模式
合同分析场景使用的是Map-Reduce模式：
- Map阶段：100份合同拆分，并行AI分析
- Reduce阶段：汇总所有分析结果，生成综合报告
二、pom.xml与配置
2.1 pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>3.3.5</version>
</parent>
<groupId>com.company</groupId>
<artifactId>agent-task-decompose</artifactId>
<version>1.0.0</version>
<properties>
<java.version>21</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<!-- Spring AI -->
<dependency>
<groupId>org.springframework.ai</groupId>
<artifactId>spring-ai-openai-spring-boot-starter</artifactId>
<version>1.0.0</version>
</dependency>
<!-- 数据库（任务状态持久化） -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-jpa</artifactId>
</dependency>
<dependency>
<groupId>com.mysql</groupId>
<artifactId>mysql-connector-j</artifactId>
<scope>runtime</scope>
</dependency>
<!-- Redis（进度追踪、分布式锁） -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<!-- Redisson（分布式锁） -->
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson-spring-boot-starter</artifactId>
<version>3.27.2</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>2.2 application.yml
spring:
application:
name: agent-task-decompose
datasource:
url: jdbc:mysql://localhost:3306/task_db?useUnicode=true&serverTimezone=Asia/Shanghai
username: ${DB_USERNAME:root}
password: ${DB_PASSWORD:password}
jpa:
hibernate:
ddl-auto: update
data:
redis:
host: localhost
port: 6379
ai:
openai:
api-key: ${OPENAI_API_KEY}
chat:
options:
model: gpt-4o
temperature: 0.1
# 任务调度配置
task:
executor:
# 并行执行线程池大小
core-pool-size: 10
max-pool-size: 20
queue-capacity: 100
thread-name-prefix: agent-task-
# 合同分析配置
contract:
# 每批处理数量
batch-size: 5
# 每批最大并发
max-concurrent: 5
# AI API速率限制（每分钟请求数）
api-rate-limit: 20
# 单个任务超时时间（秒）
single-task-timeout: 120
# 失败重试次数
max-retries: 2
server:
port: 8080三、任务模型设计
3.1 DAG任务节点
package com.company.task.model;
import jakarta.persistence.*;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.time.LocalDateTime;
import java.util.List;
/**
* 任务节点
* DAG图中的一个节点，代表一个原子任务
*/
@Entity
@Table(name = "task_node")
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class TaskNode {
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
private Long id;
@Column(name = "task_id", nullable = false, unique = true, length = 64)
private String taskId; // 任务唯一ID
@Column(name = "dag_id", nullable = false, length = 64)
private String dagId; // 所属DAG的ID
@Column(name = "task_name", nullable = false, length = 128)
private String taskName; // 任务名称（展示用）
@Column(name = "task_type", nullable = false, length = 64)
private String taskType; // 任务类型，如：CONTRACT_ANALYZE
@Column(name = "input_data", columnDefinition = "TEXT")
private String inputData; // 输入数据（JSON）
@Column(name = "output_data", columnDefinition = "TEXT")
private String outputData; // 输出数据（JSON），完成后写入
@Column(name = "status", nullable = false, length = 32)
@Enumerated(EnumType.STRING)
private TaskStatus status; // 任务状态
@Column(name = "retry_count")
private int retryCount; // 已重试次数
@Column(name = "error_message", columnDefinition = "TEXT")
private String errorMessage;
@Column(name = "depends_on", columnDefinition = "TEXT")
private String dependsOn; // 依赖的任务ID列表（JSON数组）
@Column(name = "priority")
private int priority; // 优先级，数字越小优先级越高
@Column(name = "created_at")
private LocalDateTime createdAt;
@Column(name = "started_at")
private LocalDateTime startedAt;
@Column(name = "finished_at")
private LocalDateTime finishedAt;
@PrePersist
public void prePersist() {
if (createdAt == null) createdAt = LocalDateTime.now();
if (status == null) status = TaskStatus.PENDING;
}
public boolean isReady(java.util.Set<String> completedTaskIds) {
if (dependsOn == null || dependsOn.isBlank()) return true;
// 检查所有依赖是否已完成
// 实际解析JSON，此处简化
return completedTaskIds.contains(dependsOn);
}
}package com.company.task.model;
/**
* 任务状态
*/
public enum TaskStatus {
PENDING, // 等待执行
READY, // 依赖已满足，可以执行
RUNNING, // 执行中
SUCCESS, // 执行成功
FAILED, // 执行失败
RETRYING, // 重试中
SKIPPED, // 跳过（依赖失败）
CANCELLED // 已取消
}package com.company.task.model;
import jakarta.persistence.*;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.time.LocalDateTime;
/**
* DAG任务图
* 一批合同分析任务对应一个DAG
*/
@Entity
@Table(name = "task_dag")
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class TaskDag {
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
private Long id;
@Column(name = "dag_id", nullable = false, unique = true)
private String dagId;
@Column(name = "dag_name", nullable = false)
private String dagName;
@Column(name = "status", nullable = false)
@Enumerated(EnumType.STRING)
private TaskStatus status;
@Column(name = "total_tasks")
private int totalTasks;
@Column(name = "completed_tasks")
private int completedTasks;
@Column(name = "failed_tasks")
private int failedTasks;
@Column(name = "created_by")
private String createdBy;
@Column(name = "created_at")
private LocalDateTime createdAt;
@Column(name = "started_at")
private LocalDateTime startedAt;
@Column(name = "finished_at")
private LocalDateTime finishedAt;
@Column(name = "summary_result", columnDefinition = "TEXT")
private String summaryResult; // 最终汇总结果
@PrePersist
public void prePersist() {
if (createdAt == null) createdAt = LocalDateTime.now();
if (status == null) status = TaskStatus.PENDING;
}
public double getProgressPercent() {
if (totalTasks == 0) return 0;
return (double) completedTasks / totalTasks * 100;
}
}四、Planner Agent：生成任务分解计划
package com.company.task.planner;
import com.company.task.model.TaskNode;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.ai.chat.client.ChatClient;
import org.springframework.stereotype.Service;
import java.util.List;
import java.util.UUID;
/**
* Planner Agent
* 负责将大任务分解为子任务列表
*/
@Slf4j
@Service
@RequiredArgsConstructor
public class ContractPlannerAgent {
private final ChatClient chatClient;
private final ObjectMapper objectMapper;
private static final String PLANNER_PROMPT = """
你是一个任务规划专家。
我需要分析以下%d份合同：
%s
请将这些合同按照以下规则分组，生成任务分解计划：
1. 独立合同可以并行分析
2. 补充协议必须在主合同分析后才能分析（depends_on字段）
3. 按优先级排序（关键合同优先，priority=1最高）
请返回JSON格式的任务列表：
[
{
"taskName": "分析合同-XXX",
"taskType": "CONTRACT_ANALYZE",
"inputData": {"contractId": "C001", "contractPath": "..."},
"dependsOn": null,
"priority": 1
}
]
只返回JSON，不要其他说明文字。
""";
/**
* 生成合同分析任务计划
*/
public List<TaskNode> generatePlan(List<ContractInfo> contracts, String dagId) {
log.info("生成任务分解计划: dagId={}, 合同数量={}", dagId, contracts.size());
// 构建合同列表描述
StringBuilder contractList = new StringBuilder();
for (ContractInfo contract : contracts) {
contractList.append(String.format("- ID:%s, 名称:%s, 类型:%s\n",
contract.getId(), contract.getName(), contract.getType()));
}
String prompt = String.format(PLANNER_PROMPT, contracts.size(), contractList);
String planJson = chatClient.prompt()
.user(prompt)
.call()
.content();
try {
// 提取JSON部分（防止AI返回额外说明文字）
planJson = extractJson(planJson);
List<TaskPlanItem> planItems = objectMapper.readValue(
planJson, new TypeReference<>() {});
// 转换为TaskNode列表
return planItems.stream()
.map(item -> TaskNode.builder()
.taskId(UUID.randomUUID().toString())
.dagId(dagId)
.taskName(item.getTaskName())
.taskType(item.getTaskType())
.inputData(toJson(item.getInputData()))
.dependsOn(item.getDependsOn())
.priority(item.getPriority() != null ? item.getPriority() : 5)
.build())
.toList();
} catch (Exception e) {
log.error("任务计划解析失败，使用默认策略", e);
// 降级：简单地为每份合同创建独立任务（无依赖）
return contracts.stream()
.map(contract -> TaskNode.builder()
.taskId(UUID.randomUUID().toString())
.dagId(dagId)
.taskName("分析合同-" + contract.getName())
.taskType("CONTRACT_ANALYZE")
.inputData(toJson(contract))
.priority(5)
.build())
.toList();
}
}
private String extractJson(String text) {
int start = text.indexOf('[');
int end = text.lastIndexOf(']');
if (start >= 0 && end > start) {
return text.substring(start, end + 1);
}
return text;
}
private String toJson(Object obj) {
try {
return objectMapper.writeValueAsString(obj);
} catch (Exception e) {
return "{}";
}
}
}五、并行执行框架：线程池 + CompletableFuture
5.1 线程池配置
package com.company.task.config;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.Executor;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadPoolExecutor;
/**
* 任务执行线程池配置
*/
@Configuration
public class TaskExecutorConfig {
@Value("${task.executor.core-pool-size:10}")
private int corePoolSize;
@Value("${task.executor.max-pool-size:20}")
private int maxPoolSize;
@Value("${task.executor.queue-capacity:100}")
private int queueCapacity;
@Bean("agentTaskExecutor")
public ThreadPoolTaskExecutor agentTaskExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(corePoolSize);
executor.setMaxPoolSize(maxPoolSize);
executor.setQueueCapacity(queueCapacity);
executor.setThreadNamePrefix("agent-task-");
// 拒绝策略：调用者运行，避免任务丢失
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
executor.setWaitForTasksToCompleteOnShutdown(true);
executor.setAwaitTerminationSeconds(60);
executor.initialize();
return executor;
}
/**
* AI API速率限制信号量
* 控制同时调用AI API的并发数
*/
@Bean("apiRateSemaphore")
public Semaphore apiRateSemaphore(
@Value("${task.contract.max-concurrent:5}") int maxConcurrent) {
return new Semaphore(maxConcurrent, true);
}
}5.2 DAG调度执行器
package com.company.task.executor;
import com.company.task.model.TaskDag;
import com.company.task.model.TaskNode;
import com.company.task.model.TaskStatus;
import com.company.task.repository.TaskDagRepository;
import com.company.task.repository.TaskNodeRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;
/**
* DAG任务调度执行器
* 按照依赖关系调度任务，最大化并行度
*/
@Slf4j
@Service
@RequiredArgsConstructor
public class DagTaskScheduler {
private final TaskNodeRepository nodeRepository;
private final TaskDagRepository dagRepository;
private final ContractAnalyzeExecutor contractExecutor;
private final Executor agentTaskExecutor;
private final Semaphore apiRateSemaphore;
private final TaskProgressTracker progressTracker;
/**
* 执行DAG任务图
* 自动处理依赖关系，并行执行无依赖的任务
*/
public void executeDag(String dagId) {
log.info("开始执行DAG: {}", dagId);
TaskDag dag = dagRepository.findByDagId(dagId)
.orElseThrow(() -> new IllegalArgumentException("DAG不存在: " + dagId));
dag.setStatus(TaskStatus.RUNNING);
dag.setStartedAt(LocalDateTime.now());
dagRepository.save(dag);
// 获取所有任务节点，按优先级排序
List<TaskNode> allNodes = nodeRepository.findByDagIdOrderByPriority(dagId);
Set<String> completedTaskIds = new HashSet<>();
Set<String> failedTaskIds = new HashSet<>();
// 存储所有正在执行的Future
Map<String, CompletableFuture<TaskExecutionResult>> runningFutures = new HashMap<>();
while (true) {
// 找出当前可以执行的任务（依赖已满足且未开始）
List<TaskNode> readyTasks = findReadyTasks(allNodes, completedTaskIds, failedTaskIds);
if (readyTasks.isEmpty() && runningFutures.isEmpty()) {
// 没有可执行的任务，也没有运行中的任务，结束
break;
}
// 提交可执行的任务
for (TaskNode task : readyTasks) {
if (runningFutures.containsKey(task.getTaskId())) {
continue; // 已在执行中
}
log.info("提交任务: taskId={}, taskName={}", task.getTaskId(), task.getTaskName());
CompletableFuture<TaskExecutionResult> future = CompletableFuture
.supplyAsync(() -> executeTask(task), agentTaskExecutor)
.orTimeout(120, TimeUnit.SECONDS);
runningFutures.put(task.getTaskId(), future);
// 更新任务状态为RUNNING
task.setStatus(TaskStatus.RUNNING);
task.setStartedAt(LocalDateTime.now());
nodeRepository.save(task);
}
// 等待至少一个任务完成
if (!runningFutures.isEmpty()) {
// 每500ms检查一次是否有任务完成
Iterator<Map.Entry<String, CompletableFuture<TaskExecutionResult>>> iter =
runningFutures.entrySet().iterator();
while (iter.hasNext()) {
Map.Entry<String, CompletableFuture<TaskExecutionResult>> entry = iter.next();
String taskId = entry.getKey();
CompletableFuture<TaskExecutionResult> future = entry.getValue();
if (future.isDone()) {
iter.remove();
try {
TaskExecutionResult result = future.get();
handleTaskComplete(taskId, result, completedTaskIds, failedTaskIds);
} catch (Exception e) {
handleTaskError(taskId, e, failedTaskIds);
}
// 更新DAG进度
progressTracker.updateProgress(dagId,
completedTaskIds.size(), allNodes.size());
}
}
try {
Thread.sleep(500);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
}
}
// DAG执行完成
finalizeDag(dag, completedTaskIds, failedTaskIds, allNodes.size());
}
/**
* 找出当前可以执行的任务
*/
private List<TaskNode> findReadyTasks(List<TaskNode> allNodes,
Set<String> completedTaskIds,
Set<String> failedTaskIds) {
return allNodes.stream()
.filter(node -> node.getStatus() == TaskStatus.PENDING)
.filter(node -> {
// 检查依赖是否满足
if (node.getDependsOn() == null || node.getDependsOn().isBlank()) {
return true; // 无依赖，可以执行
}
// 解析依赖的任务ID
String depId = node.getDependsOn().replaceAll("[\"\\[\\]]", "").trim();
if (failedTaskIds.contains(depId)) {
// 依赖失败，标记当前任务为SKIPPED
node.setStatus(TaskStatus.SKIPPED);
nodeRepository.save(node);
return false;
}
return completedTaskIds.contains(depId);
})
.sorted(Comparator.comparingInt(TaskNode::getPriority))
.collect(Collectors.toList());
}
/**
* 执行单个任务（在线程池中执行）
*/
private TaskExecutionResult executeTask(TaskNode task) {
try {
// 获取API调用令牌（速率限制）
apiRateSemaphore.acquire();
try {
return contractExecutor.execute(task);
} finally {
apiRateSemaphore.release();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return TaskExecutionResult.failed("任务被中断");
}
}
private void handleTaskComplete(String taskId, TaskExecutionResult result,
Set<String> completedTaskIds, Set<String> failedTaskIds) {
TaskNode node = nodeRepository.findByTaskId(taskId).orElseThrow();
if (result.isSuccess()) {
node.setStatus(TaskStatus.SUCCESS);
node.setOutputData(result.getOutputData());
completedTaskIds.add(taskId);
log.info("任务成功: taskId={}", taskId);
} else {
// 检查是否可以重试
if (node.getRetryCount() < 2) {
node.setStatus(TaskStatus.PENDING); // 重置为PENDING触发重试
node.setRetryCount(node.getRetryCount() + 1);
log.warn("任务失败，将重试: taskId={}, retryCount={}", taskId, node.getRetryCount());
} else {
node.setStatus(TaskStatus.FAILED);
node.setErrorMessage(result.getErrorMessage());
failedTaskIds.add(taskId);
log.error("任务最终失败: taskId={}", taskId);
}
}
node.setFinishedAt(LocalDateTime.now());
nodeRepository.save(node);
}
private void handleTaskError(String taskId, Exception e, Set<String> failedTaskIds) {
log.error("任务执行异常: taskId={}", taskId, e);
TaskNode node = nodeRepository.findByTaskId(taskId).orElseThrow();
node.setStatus(TaskStatus.FAILED);
node.setErrorMessage("执行异常: " + e.getMessage());
node.setFinishedAt(LocalDateTime.now());
nodeRepository.save(node);
failedTaskIds.add(taskId);
}
private void finalizeDag(TaskDag dag, Set<String> completed,
Set<String> failed, int total) {
dag.setCompletedTasks(completed.size());
dag.setFailedTasks(failed.size());
dag.setFinishedAt(LocalDateTime.now());
dag.setStatus(failed.isEmpty() ? TaskStatus.SUCCESS : TaskStatus.FAILED);
dagRepository.save(dag);
log.info("DAG执行完成: dagId={}, 成功={}/{}, 失败={}",
dag.getDagId(), completed.size(), total, failed.size());
}
}六、合同分析执行器
package com.company.task.executor;
import com.company.task.model.TaskNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.ai.chat.client.ChatClient;
import org.springframework.stereotype.Component;
import java.util.Map;
/**
* 合同分析任务执行器
* 单个合同的AI分析逻辑
*/
@Slf4j
@Component
@RequiredArgsConstructor
public class ContractAnalyzeExecutor {
private final ChatClient chatClient;
private final ObjectMapper objectMapper;
private final ContractLoader contractLoader;
private static final String ANALYZE_PROMPT = """
你是专业的劳动合同风险审查律师。
请审查以下合同内容，重点检查：
1. 试用期条款（是否超过法定上限）
2. 竞业限制条款（范围、期限、补偿金是否合法）
3. 违约金条款（金额是否过高、是否违法）
4. 工作地点和工作内容变更条款
5. 加班工资和休假条款
返回JSON格式：
{
"contractId": "合同ID",
"riskLevel": "HIGH/MEDIUM/LOW",
"issues": [
{
"clause": "条款名称",
"issue": "问题描述",
"severity": "HIGH/MEDIUM/LOW",
"suggestion": "修改建议"
}
],
"summary": "总体评价"
}
只返回JSON，不要其他说明。
""";
public TaskExecutionResult execute(TaskNode task) {
log.info("开始分析合同: taskId={}", task.getTaskId());
try {
// 解析输入数据
Map<String, Object> inputData = objectMapper.readValue(
task.getInputData(), new java.util.HashMap<String, Object>(){}.getClass());
String contractId = (String) inputData.get("contractId");
String contractPath = (String) inputData.get("contractPath");
// 加载合同内容
String contractContent = contractLoader.loadContract(contractPath);
if (contractContent == null || contractContent.isBlank()) {
return TaskExecutionResult.failed("合同内容为空: " + contractPath);
}
// 如果合同超长，截取关键部分（前8000字 + 后2000字）
if (contractContent.length() > 12000) {
contractContent = contractContent.substring(0, 8000)
+ "\n[...内容较长，已截取关键部分...]\n"
+ contractContent.substring(contractContent.length() - 2000);
}
// 调用AI分析
String analysisJson = chatClient.prompt()
.system(ANALYZE_PROMPT)
.user("合同ID：" + contractId + "\n\n合同内容：\n" + contractContent)
.call()
.content();
// 提取并验证JSON
analysisJson = extractJson(analysisJson);
// 验证JSON格式
objectMapper.readTree(analysisJson);
log.info("合同分析完成: taskId={}, contractId={}", task.getTaskId(), contractId);
return TaskExecutionResult.builder()
.success(true)
.outputData(analysisJson)
.build();
} catch (Exception e) {
log.error("合同分析失败: taskId={}", task.getTaskId(), e);
return TaskExecutionResult.failed("分析失败: " + e.getMessage());
}
}
private String extractJson(String text) {
int start = text.indexOf('{');
int end = text.lastIndexOf('}');
if (start >= 0 && end > start) {
return text.substring(start, end + 1);
}
return text;
}
}七、结果聚合：Reduce阶段
package com.company.task.aggregator;
import com.company.task.model.TaskDag;
import com.company.task.model.TaskNode;
import com.company.task.model.TaskStatus;
import com.company.task.repository.TaskNodeRepository;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.ai.chat.client.ChatClient;
import org.springframework.stereotype.Service;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
/**
* 结果聚合服务
* 将多个子任务的结果合并成最终报告
*/
@Slf4j
@Service
@RequiredArgsConstructor
public class ContractAnalysisAggregator {
private final TaskNodeRepository nodeRepository;
private final ChatClient chatClient;
private final ObjectMapper objectMapper;
/**
* 聚合所有合同分析结果，生成综合报告
*/
public String aggregate(String dagId) {
log.info("开始聚合分析结果: dagId={}", dagId);
// 获取所有成功完成的任务结果
List<TaskNode> successNodes = nodeRepository.findByDagIdAndStatus(
dagId, TaskStatus.SUCCESS);
List<TaskNode> failedNodes = nodeRepository.findByDagIdAndStatus(
dagId, TaskStatus.FAILED);
if (successNodes.isEmpty()) {
return buildEmptyReport(failedNodes.size());
}
// 解析各合同分析结果
List<Map<String, Object>> analysisResults = new ArrayList<>();
for (TaskNode node : successNodes) {
try {
Map<String, Object> result = objectMapper.readValue(
node.getOutputData(), new java.util.HashMap<String, Object>(){}.getClass());
analysisResults.add(result);
} catch (Exception e) {
log.warn("解析任务结果失败: taskId={}", node.getTaskId(), e);
}
}
// 统计风险分布
Map<String, Long> riskDistribution = analysisResults.stream()
.collect(Collectors.groupingBy(
r -> String.valueOf(r.getOrDefault("riskLevel", "UNKNOWN")),
Collectors.counting()));
// 提取所有高风险问题
List<Map<String, Object>> highRiskIssues = analysisResults.stream()
.filter(r -> "HIGH".equals(r.get("riskLevel")))
.limit(20) // 最多展示20个高风险合同
.toList();
// 生成综合报告（用AI总结）
String summaryPrompt = buildSummaryPrompt(analysisResults, riskDistribution, failedNodes.size());
String summary = chatClient.prompt()
.user(summaryPrompt)
.call()
.content();
// 构建最终报告
return buildFinalReport(dagId, analysisResults.size(), failedNodes.size(),
riskDistribution, highRiskIssues, summary);
}
private String buildSummaryPrompt(List<Map<String, Object>> results,
Map<String, Long> distribution,
int failedCount) {
return String.format("""
我对%d份劳动合同进行了风险审查，结果如下：
- 高风险合同：%d份
- 中风险合同：%d份
- 低风险合同：%d份
- 分析失败：%d份
高风险合同中最常见的问题包括：竞业限制过宽、违约金过高、试用期超标。
请用200字内总结整体风险状况，并给出3条最重要的改进建议。
""",
results.size(),
distribution.getOrDefault("HIGH", 0L),
distribution.getOrDefault("MEDIUM", 0L),
distribution.getOrDefault("LOW", 0L),
failedCount);
}
private String buildFinalReport(String dagId, int successCount, int failedCount,
Map<String, Long> riskDistribution,
List<Map<String, Object>> highRiskIssues,
String summary) {
try {
Map<String, Object> report = Map.of(
"dagId", dagId,
"totalAnalyzed", successCount,
"failedCount", failedCount,
"riskDistribution", riskDistribution,
"highRiskContractsCount", riskDistribution.getOrDefault("HIGH", 0L),
"topHighRiskIssues", highRiskIssues,
"executiveSummary", summary,
"generatedAt", java.time.LocalDateTime.now().toString()
);
return objectMapper.writeValueAsString(report);
} catch (Exception e) {
return "{\"error\": \"报告生成失败\"}";
}
}
private String buildEmptyReport(int failedCount) {
return String.format("{\"error\": \"所有任务失败，无有效结果\", \"failedCount\": %d}", failedCount);
}
}八、进度追踪：实时展示大任务进度
8.1 进度追踪服务
package com.company.task.progress;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Map;
/**
* 任务进度追踪服务
* 使用Redis存储实时进度，支持SSE推送
*/
@Slf4j
@Service
@RequiredArgsConstructor
public class TaskProgressTracker {
private final StringRedisTemplate redisTemplate;
private static final String PROGRESS_PREFIX = "task:progress:";
private static final Duration PROGRESS_TTL = Duration.ofHours(24);
/**
* 更新任务进度
*/
public void updateProgress(String dagId, int completedCount, int totalCount) {
String key = PROGRESS_PREFIX + dagId;
Map<Object, Object> progressData = Map.of(
"dagId", dagId,
"completed", String.valueOf(completedCount),
"total", String.valueOf(totalCount),
"percent", String.format("%.1f", (double) completedCount / totalCount * 100),
"updatedAt", LocalDateTime.now().toString()
);
redisTemplate.opsForHash().putAll(key, progressData);
redisTemplate.expire(key, PROGRESS_TTL);
log.debug("进度更新: dagId={}, {}/{}", dagId, completedCount, totalCount);
}
/**
* 获取任务进度
*/
public TaskProgress getProgress(String dagId) {
String key = PROGRESS_PREFIX + dagId;
Map<Object, Object> data = redisTemplate.opsForHash().entries(key);
if (data.isEmpty()) {
return TaskProgress.notFound(dagId);
}
return TaskProgress.builder()
.dagId(dagId)
.completed(parseIntSafe(data.get("completed")))
.total(parseIntSafe(data.get("total")))
.percent(parseDoubleSafe(data.get("percent")))
.build();
}
private int parseIntSafe(Object val) {
try {
return Integer.parseInt(String.valueOf(val));
} catch (Exception e) {
return 0;
}
}
private double parseDoubleSafe(Object val) {
try {
return Double.parseDouble(String.valueOf(val));
} catch (Exception e) {
return 0.0;
}
}
}8.2 SSE进度推送接口
package com.company.task.controller;
import com.company.task.progress.TaskProgress;
import com.company.task.progress.TaskProgressTracker;
import lombok.RequiredArgsConstructor;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.io.IOException;
import java.util.concurrent.Executors;
/**
* 任务进度SSE接口
* 客户端通过EventSource订阅实时进度
*/
@RestController
@RequestMapping("/api/tasks")
@RequiredArgsConstructor
public class TaskProgressController {
private final TaskProgressTracker progressTracker;
/**
* SSE进度推送
* 前端：const es = new EventSource('/api/tasks/dagId123/progress/stream')
*/
@GetMapping(value = "/{dagId}/progress/stream",
produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter streamProgress(@PathVariable String dagId) {
SseEmitter emitter = new SseEmitter(300_000L); // 5分钟超时
Executors.newSingleThreadExecutor().execute(() -> {
try {
while (true) {
TaskProgress progress = progressTracker.getProgress(dagId);
emitter.send(SseEmitter.event()
.name("progress")
.data(progress));
// 完成时关闭连接
if (progress.isCompleted()) {
emitter.complete();
break;
}
Thread.sleep(1000); // 每秒推送一次
}
} catch (IOException | InterruptedException e) {
emitter.completeWithError(e);
}
});
return emitter;
}
/**
* 一次性查询进度
*/
@GetMapping("/{dagId}/progress")
public TaskProgress getProgress(@PathVariable String dagId) {
return progressTracker.getProgress(dagId);
}
}九、对外API接口
package com.company.task.controller;
import com.company.task.model.TaskDag;
import com.company.task.service.ContractAnalysisService;
import lombok.RequiredArgsConstructor;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;
import java.util.List;
import java.util.Map;
/**
* 合同分析任务API
*/
@RestController
@RequestMapping("/api/contract-analysis")
@RequiredArgsConstructor
public class ContractAnalysisController {
private final ContractAnalysisService analysisService;
/**
* 提交批量合同分析任务
*/
@PostMapping("/submit")
public ResponseEntity<Map<String, String>> submitBatch(
@RequestParam("files") List<MultipartFile> files,
@RequestParam("submittedBy") String submittedBy) {
String dagId = analysisService.submitBatch(files, submittedBy);
return ResponseEntity.ok(Map.of(
"dagId", dagId,
"message", String.format("已提交%d份合同，正在异步分析", files.size()),
"progressUrl", "/api/tasks/" + dagId + "/progress"
));
}
/**
* 获取分析报告
*/
@GetMapping("/{dagId}/report")
public ResponseEntity<String> getReport(@PathVariable String dagId) {
String report = analysisService.getReport(dagId);
if (report == null) {
return ResponseEntity.notFound().build();
}
return ResponseEntity.ok()
.header("Content-Type", "application/json; charset=UTF-8")
.body(report);
}
/**
* 获取任务详情（各合同分析状态）
*/
@GetMapping("/{dagId}/tasks")
public ResponseEntity<List<Map<String, Object>>> getTaskDetails(@PathVariable String dagId) {
return ResponseEntity.ok(analysisService.getTaskDetails(dagId));
}
}十、性能数据对比
实际测试数据（100份合同，平均每份8000字）：
| 方案 | 执行时间 | API调用次数 | 失败率 | 备注 |
|---|---|---|---|---|
| 串行处理 | 118分钟 | 100次 | 3% | 原始方案 |
| 简单并发（无限流） | 失败 | 100次 | 100% | 触发Rate Limit |
| 5并发批处理 | 24分钟 | 100次 | 2% | 受API限制 |
| DAG调度+优先级+重试 | 8分钟 | 106次（含重试） | 0% | 最终方案 |
8分钟 vs 118分钟，效率提升14.75倍。
关键性能指标：
- 平均单合同分析时间：4.8秒（包括AI调用、解析、存储）
- 最大并发AI调用：5（受API Rate Limit约束）
- 任务调度开销：< 10ms/任务
- 数据库写入：异步批量写入，不影响主流程
FAQ
Q1：如果某个子任务一直失败，整个DAG会卡住吗？
不会。每个任务有最大重试次数（默认2次），超过后标记为FAILED。依赖失败任务的后续任务标记为SKIPPED。DAG会在所有任务都到达终态（SUCCESS/FAILED/SKIPPED）后结束。
Q2：如何处理合同超长（比如100页）的情况？
本文方案是截取前8000字+后2000字。更好的方案是：将长合同分块，每块单独分析，然后再做二次合并。这样不丢失中间内容，但会增加AI调用次数。根据业务重要性选择。
Q3：Planner Agent生成的任务计划不准确怎么办？
Planner Agent的结果只是建议，核心是保底策略：如果Planner失败，退化为"每份合同独立并行"的简单策略。真实业务中，可以让Planner只处理依赖关系推断（这是它擅长的），任务分片逻辑用硬编码规则处理（更可靠）。
Q4：分布式环境下多个实例会重复执行任务吗？
会。本文代码是单实例版本。分布式场景需要加分布式锁（Redis/Redisson）：每个任务执行前获取锁，执行完释放。任务状态用数据库乐观锁保证一致性。
Q5：DAG任务图如何持久化和恢复？
本文已经将TaskNode和TaskDag存入MySQL。如果中途服务重启，只需查询状态为PENDING/RUNNING的任务重新调度即可。RUNNING状态的任务在恢复时重置为PENDING触发重试。