一、引言：从模型训练到生产部署的挑战
将AI模型从实验环境部署到生产环境面临诸多挑战：模型版本管理、推理性能、资源隔离、流量控制、监控报警等。一个成熟的模型服务平台需要解决以下问题：

版本管理：跟踪模型版本与元数据，确保可重现性

服务化：将模型封装为高可用的API服务

流量管理：支持金丝雀发布、A/B测试和流量切换

监控报警：实时监控模型性能与数据分布变化

资源优化：高效利用GPU/CPU资源，控制成本

Java在企业级应用开发中的成熟生态，结合云原生技术，为构建此类平台提供了理想基础。

二、平台架构设计

系统架构概览

text
模型仓库 → 模型服务 → 流量管理 → 监控报警
↓ ↓ ↓ ↓
MLflow → Spring Boot → Istio → Prometheus
↓ ↓ ↓ ↓
版本控制 REST API 路由规则指标收集

核心组件

模型仓库：MLflow Model Registry

模型服务：Spring Boot + Deep Java Library (DJL)

流量管理：Istio VirtualService

监控报警：Micrometer + Prometheus + Grafana

资源管理：Kubernetes + Custom Resource Definitions (CRDs)

项目依赖配置

xml

3.2.0
0.25.0
2.8.0

org.springframework.boot
spring-boot-starter-web
${spring-boot.version}

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-actuator</artifactId>
    <version>${spring-boot.version}</version>
</dependency>

<!-- 模型推理 -->
<dependency>
    <groupId>ai.djl</groupId>
    <artifactId>api</artifactId>
    <version>${djl.version}</version>
</dependency>

<dependency>
    <groupId>ai.djl.pytorch</groupId>
    <artifactId>pytorch-engine</artifactId>
    <version>${djl.version}</version>
</dependency>

<!-- 模型仓库客户端 -->
<dependency>
    <groupId>org.mlflow</groupId>
    <artifactId>mlflow-client</artifactId>
    <version>${mlflow.version}</version>
</dependency>

<!-- 监控 -->
<dependency>
    <groupId>io.micrometer</groupId>
    <artifactId>micrometer-registry-prometheus</artifactId>
</dependency>

<!-- Kubernetes客户端 -->
<dependency>
    <groupId>io.kubernetes</groupId>
    <artifactId>client-java</artifactId>
    <version>18.0.0</version>
</dependency>

三、模型仓库与版本管理

MLflow模型注册集成

java
// MlflowModelRegistry.java
@Component
@Slf4j
public class MlflowModelRegistry {

private final MlflowClient mlflowClient;
private final String trackingUri;

public MlflowModelRegistry(@Value("${mlflow.tracking-uri}") String trackingUri) {
    this.trackingUri = trackingUri;
    this.mlflowClient = new MlflowClient(trackingUri);
}

/**
 * 注册新模型版本
 */
public String registerModelVersion(String modelName, String source, String runId) {
    try {
        return mlflowClient.createModelVersion(modelName, source, runId);
    } catch (Exception e) {
        log.error("模型版本注册失败: {}", modelName, e);
        throw new ModelRegistryException("模型注册失败", e);
    }
}

/**
 * 获取模型版本信息
 */
public ModelVersion getModelVersion(String modelName, String version) {
    try {
        return mlflowClient.getModelVersion(modelName, version);
    } catch (Exception e) {
        log.error("获取模型版本失败: {}:{}", modelName, version, e);
        throw new ModelRegistryException("获取模型版本失败", e);
    }
}

/**
 * 过渡模型版本阶段
 */
public void transitionModelVersionStage(String modelName, String version, 
                                       ModelVersion.Stage stage) {
    try {
        mlflowClient.transitionModelVersionStage(modelName, version, stage);
        log.info("模型版本阶段过渡: {}:{} -> {}", modelName, version, stage);
    } catch (Exception e) {
        log.error("模型版本阶段过渡失败: {}:{}", modelName, version, e);
        throw new ModelRegistryException("模型阶段过渡失败", e);
    }
}

/**
 * 获取生产就绪的模型版本
 */
public ModelVersion getProductionModel(String modelName) {
    try {
        // 获取所有版本并过滤出生产版本
        List<ModelVersion> versions = mlflowClient.getModelVersions(modelName);
        return versions.stream()
                .filter(v -> v.getCurrentStage() == ModelVersion.Stage.PRODUCTION)
                .max(Comparator.comparing(ModelVersion::getVersion))
                .orElseThrow(() -> new ModelNotFoundException("未找到生产模型: " + modelName));
    } catch (Exception e) {
        log.error("获取生产模型失败: {}", modelName, e);
        throw new ModelRegistryException("获取生产模型失败", e);
    }
}

/**
 * 获取模型URI用于加载
 */
public String getModelUri(String modelName, String version) {
    ModelVersion modelVersion = getModelVersion(modelName, version);
    return modelVersion.getSource();
}

}

模型元数据管理

java
// ModelMetadata.java
@Data
public class ModelMetadata {
private String name;
private String version;
private ModelVersion.Stage stage;
private String description;
private Map tags;
private String framework;
private String inputSchema;
private String outputSchema;
private Date createdAt;
private Date updatedAt;

// 从MLflow模型版本转换
public static ModelMetadata fromMlflowModelVersion(ModelVersion mv) {
    ModelMetadata metadata = new ModelMetadata();
    metadata.setName(mv.getName());
    metadata.setVersion(mv.getVersion());
    metadata.setStage(mv.getCurrentStage());
    metadata.setDescription(mv.getDescription());
    metadata.setTags(mv.getTags());
    metadata.setCreatedAt(new Date(mv.getCreationTimestamp()));
    metadata.setUpdatedAt(new Date(mv.getLastUpdatedTimestamp()));

    // 从run中获取更多元数据
    if (mv.getRun() != null) {
        metadata.setFramework(mv.getRun().getData().getTags().get("framework"));
    }

    return metadata;
}

}
四、模型服务化与推理引擎

统一模型服务接口

java
// ModelService.java
public interface ModelService {
ModelMetadata getMetadata();
InferenceResult predict(InferenceRequest request);
BatchInferenceResult batchPredict(List requests);
Health health();
void close();
}

// 基础模型服务实现
@Component
@Slf4j
public class BaseModelService implements ModelService {

private final ModelMetadata metadata;
private final Predictor predictor;
private final ModelMetrics metrics;

public BaseModelService(ModelMetadata metadata, String modelUri) {
    this.metadata = metadata;
    this.predictor = loadModel(modelUri);
    this.metrics = new ModelMetrics(metadata.getName(), metadata.getVersion());
}

@Override
public ModelMetadata getMetadata() {
    return metadata;
}

@Override
public InferenceResult predict(InferenceRequest request) {
    long startTime = System.currentTimeMillis();
    try {
        // 预处理输入
        NDList input = preprocess(request.getInput());

        // 执行推理
        NDList output = predictor.predict(input);

        // 后处理输出
        Object result = postprocess(output);

        long latency = System.currentTimeMillis() - startTime;
        metrics.recordInference(latency, true);

        return InferenceResult.success(result, latency);

    } catch (Exception e) {
        long latency = System.currentTimeMillis() - startTime;
        metrics.recordInference(latency, false);
        log.error("推理失败: {}:{}", metadata.getName(), metadata.getVersion(), e);
        return InferenceResult.error(e.getMessage());
    }
}

@Override
public BatchInferenceResult batchPredict(List<InferenceRequest> requests) {
    List<InferenceResult> results = new ArrayList<>();
    for (InferenceRequest request : requests) {
        results.add(predict(request));
    }
    return new BatchInferenceResult(results);
}

@Override
public Health health() {
    try {
        // 执行一次简单的推理来检查模型健康状态
        NDList testInput = createTestInput();
        predictor.predict(testInput);
        return Health.up().build();
    } catch (Exception e) {
        return Health.down(e).build();
    }
}

@Override
public void close() {
    if (predictor != null) {
        predictor.close();
    }
}

private Predictor loadModel(String modelUri) {
    try {
        Criteria<NDList, NDList> criteria = Criteria.builder()
                .setTypes(NDList.class, NDList.class)
                .optModelUrls(modelUri)
                .build();

        ZooModel<NDList, NDList> model = criteria.loadModel();
        return model.newPredictor();

    } catch (Exception e) {
        log.error("模型加载失败: {}", modelUri, e);
        throw new ModelLoadingException("模型加载失败", e);
    }
}

private NDList preprocess(Object input) {
    // 根据模型输入模式预处理输入数据
    // 实现取决于具体模型
    return null; // 简化实现
}

private Object postprocess(NDList output) {
    // 后处理模型输出
    return null; // 简化实现
}

private NDList createTestInput() {
    // 创建测试输入
    return null; // 简化实现
}

}

模型服务管理器

java
// ModelServiceManager.java
@Component
@Slf4j
public class ModelServiceManager {

private final Map<String, ModelService> modelServices;
private final MlflowModelRegistry modelRegistry;
private final ModelServiceFactory modelServiceFactory;

public ModelServiceManager(MlflowModelRegistry modelRegistry,
                          ModelServiceFactory modelServiceFactory) {
    this.modelRegistry = modelRegistry;
    this.modelServiceFactory = modelServiceFactory;
    this.modelServices = new ConcurrentHashMap<>();
}

/**
 * 加载模型服务
 */
public ModelService loadModel(String modelName, String version) {
    String key = generateKey(modelName, version);

    return modelServices.computeIfAbsent(key, k -> {
        try {
            log.info("加载模型服务: {}:{}", modelName, version);

            // 从模型仓库获取模型URI
            String modelUri = modelRegistry.getModelUri(modelName, version);

            // 获取模型元数据
            ModelVersion mv = modelRegistry.getModelVersion(modelName, version);
            ModelMetadata metadata = ModelMetadata.fromMlflowModelVersion(mv);

            // 创建模型服务
            return modelServiceFactory.createService(metadata, modelUri);

        } catch (Exception e) {
            log.error("加载模型服务失败: {}:{}", modelName, version, e);
            throw new ModelLoadingException("模型服务加载失败", e);
        }
    });
}

/**
 * 卸载模型服务
 */
public void unloadModel(String modelName, String version) {
    String key = generateKey(modelName, version);
    ModelService service = modelServices.remove(key);
    if (service != null) {
        log.info("卸载模型服务: {}:{}", modelName, version);
        service.close();
    }
}

/**
 * 获取模型服务
 */
public ModelService getModelService(String modelName, String version) {
    String key = generateKey(modelName, version);
    ModelService service = modelServices.get(key);
    if (service == null) {
        throw new ModelNotFoundException("模型服务未加载: " + key);
    }
    return service;
}

/**
 * 获取所有已加载的模型服务
 */
public List<ModelMetadata> getLoadedModels() {
    return modelServices.values().stream()
            .map(ModelService::getMetadata)
            .collect(Collectors.toList());
}

/**
 * 自动加载生产模型
 */
@EventListener(ApplicationReadyEvent.class)
public void autoLoadProductionModels() {
    try {
        // 从配置中获取需要自动加载的模型列表
        List<String> modelsToLoad = getModelsToAutoLoad();

        for (String modelName : modelsToLoad) {
            try {
                ModelVersion productionModel = modelRegistry.getProductionModel(modelName);
                loadModel(modelName, productionModel.getVersion());
            } catch (Exception e) {
                log.warn("自动加载生产模型失败: {}", modelName, e);
            }
        }
    } catch (Exception e) {
        log.error("自动加载生产模型失败", e);
    }
}

private String generateKey(String modelName, String version) {
    return modelName + ":" + version;
}

private List<String> getModelsToAutoLoad() {
    // 从配置中获取需要自动加载的模型列表
    return List.of("sentiment-analysis", "image-classification");
}

}
五、流量管理与A/B测试

推理路由服务

java
// InferenceRouter.java
@Component
@Slf4j
public class InferenceRouter {

private final ModelServiceManager modelServiceManager;
private final RoutingRuleManager routingRuleManager;
private final TrafficSplitter trafficSplitter;

public InferenceRouter(ModelServiceManager modelServiceManager,
                      RoutingRuleManager routingRuleManager,
                      TrafficSplitter trafficSplitter) {
    this.modelServiceManager = modelServiceManager;
    this.routingRuleManager = routingRuleManager;
    this.trafficSplitter = trafficSplitter;
}

/**
 * 路由推理请求
 */
public InferenceResult route(String modelName, InferenceRequest request) {
    // 获取当前路由规则
    RoutingRule rule = routingRuleManager.getRule(modelName);

    // 根据路由规则选择模型版本
    String selectedVersion = selectModelVersion(modelName, request, rule);

    // 获取模型服务并执行推理
    ModelService modelService = modelServiceManager.getModelService(modelName, selectedVersion);
    return modelService.predict(request);
}

/**
 * 批量路由推理请求
 */
public BatchInferenceResult batchRoute(String modelName, List<InferenceRequest> requests) {
    // 为每个请求单独路由，以支持更细粒度的流量控制
    List<InferenceResult> results = new ArrayList<>();
    for (InferenceRequest request : requests) {
        results.add(route(modelName, request));
    }
    return new BatchInferenceResult(results);
}

private String selectModelVersion(String modelName, InferenceRequest request, RoutingRule rule) {
    if (rule == null) {
        // 默认返回生产版本
        return getDefaultProductionVersion(modelName);
    }

    // 检查基于内容的路由
    String contentBasedVersion = rule.getContentBasedVersion(request);
    if (contentBasedVersion != null) {
        return contentBasedVersion;
    }

    // 流量拆分
    return trafficSplitter.split(modelName, request, rule.getSplits());
}

private String getDefaultProductionVersion(String modelName) {
    // 从模型服务管理器中获取生产版本
    // 简化实现
    return "1";
}

}

// 路由规则管理
@Component
public class RoutingRuleManager {

private final Map<String, RoutingRule> rules;

public RoutingRuleManager() {
    this.rules = new ConcurrentHashMap<>();
}

public RoutingRule getRule(String modelName) {
    return rules.get(modelName);
}

public void setRule(String modelName, RoutingRule rule) {
    rules.put(modelName, rule);
    log.info("设置路由规则: {} -> {}", modelName, rule);
}

public void removeRule(String modelName) {
    rules.remove(modelName);
    log.info("移除路由规则: {}", modelName);
}

}

// 流量拆分器
@Component
public class TrafficSplitter {

private final Random random;

public TrafficSplitter() {
    this.random = new Random();
}

public String split(String modelName, InferenceRequest request, Map<String, Double> splits) {
    double value = random.nextDouble();
    double cumulative = 0.0;

    for (Map.Entry<String, Double> entry : splits.entrySet()) {
        cumulative += entry.getValue();
        if (value <= cumulative) {
            return entry.getKey();
        }
    }

    // 默认返回第一个版本
    return splits.keySet().iterator().next();
}

}

// 路由规则类
@Data
public class RoutingRule {
private String modelName;
private Map splits; // 版本 -> 流量比例
private List contentBasedRules;

public String getContentBasedVersion(InferenceRequest request) {
    if (contentBasedRules == null) {
        return null;
    }

    for (ContentBasedRule rule : contentBasedRules) {
        if (rule.matches(request)) {
            return rule.getVersion();
        }
    }

    return null;
}

}

@Data
public class ContentBasedRule {
private String field;
private Object value;
private String operator; // "equals", "contains", "greater_than", etc.
private String version;

public boolean matches(InferenceRequest request) {
    // 根据字段和操作符匹配请求
    // 简化实现
    return false;
}

}
六、监控与可观测性

模型性能监控

java
// ModelMetrics.java
@Component
@Slf4j
public class ModelMetrics {

private final MeterRegistry meterRegistry;
private final Map<String, Timer> inferenceTimers;
private final Map<String, Counter> inferenceCounters;
private final Map<String, Counter> errorCounters;

public ModelMetrics(MeterRegistry meterRegistry) {
    this.meterRegistry = meterRegistry;
    this.inferenceTimers = new ConcurrentHashMap<>();
    this.inferenceCounters = new ConcurrentHashMap<>();
    this.errorCounters = new ConcurrentHashMap<>();
}

public void recordInference(String modelName, String version, long latency, boolean success) {
    String key = modelName + ":" + version;

    // 记录推理延迟
    Timer timer = inferenceTimers.computeIfAbsent(key, k -> 
        Timer.builder("model.inference.duration")
            .tag("model", modelName)
            .tag("version", version)
            .register(meterRegistry)
    );
    timer.record(latency, TimeUnit.MILLISECONDS);

    // 记录推理计数
    Counter counter = inferenceCounters.computeIfAbsent(key, k -> 
        Counter.builder("model.inference.count")
            .tag("model", modelName)
            .tag("version", version)
            .tag("success", String.valueOf(success))
            .register(meterRegistry)
    );
    counter.increment();

    if (!success) {
        // 记录错误计数
        Counter errorCounter = errorCounters.computeIfAbsent(key, k -> 
            Counter.builder("model.inference.errors")
                .tag("model", modelName)
                .tag("version", version)
                .register(meterRegistry)
        );
        errorCounter.increment();
    }
}

public void recordDataDrift(String modelName, String version, double driftScore) {
    Gauge.builder("model.data.drift")
        .tag("model", modelName)
        .tag("version", version)
        .register(meterRegistry, () -> driftScore);
}

public void recordPredictionDistribution(String modelName, String version, 
                                       String classLabel, double confidence) {
    // 记录预测分布
    DistributionSummary summary = DistributionSummary.builder("model.prediction.confidence")
        .tag("model", modelName)
        .tag("version", version)
        .tag("class", classLabel)
        .register(meterRegistry);
    summary.record(confidence);
}

}

// 模型健康检查
@Component
public class ModelHealthIndicator implements HealthIndicator {

private final ModelServiceManager modelServiceManager;

public ModelHealthIndicator(ModelServiceManager modelServiceManager) {
    this.modelServiceManager = modelServiceManager;
}

@Override
public Health health() {
    Map<String, Object> details = new HashMap<>();
    List<ModelMetadata> loadedModels = modelServiceManager.getLoadedModels();

    boolean allHealthy = true;
    for (ModelMetadata model : loadedModels) {
        try {
            ModelService service = modelServiceManager.getModelService(
                model.getName(), model.getVersion());
            Health serviceHealth = service.health();
            details.put(model.getName() + ":" + model.getVersion(), serviceHealth.getStatus());

            if (serviceHealth.getStatus() != Status.UP) {
                allHealthy = false;
            }
        } catch (Exception e) {
            details.put(model.getName() + ":" + model.getVersion(), "ERROR: " + e.getMessage());
            allHealthy = false;
        }
    }

    if (allHealthy) {
        return Health.up().withDetails(details).build();
    } else {
        return Health.down().withDetails(details).build();
    }
}

}
七、 REST API与模型管理

模型推理API

java
// ModelInferenceController.java
@RestController
@RequestMapping("/api/models")
@Slf4j
public class ModelInferenceController {

private final InferenceRouter inferenceRouter;

public ModelInferenceController(InferenceRouter inferenceRouter) {
    this.inferenceRouter = inferenceRouter;
}

@PostMapping("/{modelName}/predict")
public ResponseEntity<InferenceResult> predict(
        @PathVariable String modelName,
        @RequestBody InferenceRequest request) {

    try {
        InferenceResult result = inferenceRouter.route(modelName, request);
        return ResponseEntity.ok(result);
    } catch (ModelNotFoundException e) {
        return ResponseEntity.notFound().build();
    } catch (Exception e) {
        log.error("推理请求处理失败: {}", modelName, e);
        return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                .body(InferenceResult.error(e.getMessage()));
    }
}

@PostMapping("/{modelName}/batch-predict")
public ResponseEntity<BatchInferenceResult> batchPredict(
        @PathVariable String modelName,
        @RequestBody List<InferenceRequest> requests) {

    try {
        BatchInferenceResult result = inferenceRouter.batchRoute(modelName, requests);
        return ResponseEntity.ok(result);
    } catch (ModelNotFoundException e) {
        return ResponseEntity.notFound().build();
    } catch (Exception e) {
        log.error("批量推理请求处理失败: {}", modelName, e);
        return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
    }
}

}

// 模型管理API
@RestController
@RequestMapping("/api/model-management")
@Slf4j
public class ModelManagementController {

private final ModelServiceManager modelServiceManager;
private final RoutingRuleManager routingRuleManager;
private final MlflowModelRegistry modelRegistry;

public ModelManagementController(ModelServiceManager modelServiceManager,
                                RoutingRuleManager routingRuleManager,
                                MlflowModelRegistry modelRegistry) {
    this.modelServiceManager = modelServiceManager;
    this.routingRuleManager = routingRuleManager;
    this.modelRegistry = modelRegistry;
}

@PostMapping("/{modelName}/load")
public ResponseEntity<String> loadModel(
        @PathVariable String modelName,
        @RequestParam String version) {

    try {
        modelServiceManager.loadModel(modelName, version);
        return ResponseEntity.ok("模型加载成功");
    } catch (Exception e) {
        log.error("模型加载失败: {}:{}", modelName, version, e);
        return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                .body("模型加载失败: " + e.getMessage());
    }
}

@PostMapping("/{modelName}/unload")
public ResponseEntity<String> unloadModel(
        @PathVariable String modelName,
        @RequestParam String version) {

    try {
        modelServiceManager.unloadModel(modelName, version);
        return ResponseEntity.ok("模型卸载成功");
    } catch (Exception e) {
        log.error("模型卸载失败: {}:{}", modelName, version, e);
        return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                .body("模型卸载失败: " + e.getMessage());
    }
}

@GetMapping("/loaded-models")
public ResponseEntity<List<ModelMetadata>> getLoadedModels() {
    List<ModelMetadata> models = modelServiceManager.getLoadedModels();
    return ResponseEntity.ok(models);
}

@PostMapping("/{modelName}/routing-rule")
public ResponseEntity<String> setRoutingRule(
        @PathVariable String modelName,
        @RequestBody RoutingRule rule) {

    try {
        routingRuleManager.setRule(modelName, rule);
        return ResponseEntity.ok("路由规则设置成功");
    } catch (Exception e) {
        log.error("路由规则设置失败: {}", modelName, e);
        return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                .body("路由规则设置失败: " + e.getMessage());
    }
}

@PostMapping("/{modelName}/promote")
public ResponseEntity<String> promoteToProduction(
        @PathVariable String modelName,
        @RequestParam String version) {

    try {
        modelRegistry.transitionModelVersionStage(modelName, version, 
                                                 ModelVersion.Stage.PRODUCTION);
        return ResponseEntity.ok("模型已升级到生产环境");
    } catch (Exception e) {
        log.error("模型升级失败: {}:{}", modelName, version, e);
        return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                .body("模型升级失败: " + e.getMessage());
    }
}

}
八、生产配置与部署

Kubernetes部署配置

yaml

deployment.yaml

apiVersion: apps/v1
kind: Deployment
metadata:
name: model-service
spec:
replicas: 3
selector:
matchLabels:
app: model-service
template:
metadata:
labels:
app: model-service
spec:
containers:

  - name: model-service
    image: my-registry/model-service:1.0.0
    ports:
    - containerPort: 8080
    env:
    - name: MLFLOW_TRACKING_URI
      value: "http://mlflow-server:5000"
    - name: SPRING_PROFILES_ACTIVE
      value: "production"
    resources:
      requests:
        memory: "2Gi"
        cpu: "1000m"
      limits:
        memory: "4Gi"
        cpu: "2000m"
    livenessProbe:
      httpGet:
        path: /actuator/health
        port: 8080
      initialDelaySeconds: 60
      periodSeconds: 30
    readinessProbe:
      httpGet:
        path: /actuator/health
        port: 8080
      initialDelaySeconds: 30
      periodSeconds: 20

apiVersion: v1
kind: Service
metadata:
name: model-service
spec:
selector:
app: model-service
ports:

port: 80
targetPort: 8080
1. 应用配置

yaml

application-production.yml

server:
port: 8080

management:
endpoints:
web:
exposure:
include: health,metrics,info,prometheus
endpoint:
health:
show-details: always
metrics:
export:
prometheus:
enabled: true

mlflow:
tracking-uri: http://mlflow-server:5000

model:
auto-load:
enabled: true
models:

  - sentiment-analysis
  - image-classification

logging:
level:
com.example.modelservice: INFO
file:
name: /var/log/model-service.log
九、总结
通过本文的实践，我们构建了一个完整的企业级模型服务平台，具备以下核心能力：

集中式模型管理：通过MLflow集成实现模型版本控制和生命周期管理

高性能推理服务：基于DJL提供统一推理接口，支持多种深度学习框架

智能流量路由：支持A/B测试、金丝雀发布和基于内容的路由

全面监控报警：集成Micrometer和Prometheus，实时监控模型性能

云原生部署：基于Kubernetes实现高可用和弹性伸缩

这种架构使得数据科学家能够专注于模型开发，而运维团队则可以高效管理生产环境中的模型服务，为企业AI应用提供稳定可靠的推理能力。随着MLOps实践的不断成熟，这种模型服务平台将成为企业AI基础设施的核心组件。

标题：Java与强化学习：构建自适应决策与智能控制系统
摘要：强化学习作为AI领域的关键分支，使系统能够通过与环境交互自主学习最优决策策略。本文深入探讨如何在Java生态中构建基于强化学习的自适应决策系统，涵盖从环境建模、算法实现到实时控制的全流程。我们将完整展示Q学习、深度Q网络、策略梯度等核心算法在Java中的实现，以及如何将其应用于机器人控制、游戏AI、资源调度等现实场景，为构建真正自学习和自适应的智能系统提供完整技术方案。

文章内容
一、引言：从静态规则到自适应学习的进化
传统基于规则的决策系统在面对复杂、动态的环境时显得力不从心。强化学习通过"试错-奖励"机制，使系统能够：

自主学习：无需大量标注数据，通过交互学习最优策略

动态适应：实时调整策略以适应环境变化

长期优化：考虑决策的长期影响而非即时收益

泛化能力：将学到的策略迁移到类似场景

Java在实时系统、企业应用和大规模分布式计算中的优势，使其成为构建生产级强化学习系统的理想平台。本文将基于Deep Java Library、ND4J和自定义RL框架，演示如何构建高效、可扩展的强化学习解决方案。

二、强化学习核心架构设计

系统架构概览

text
环境模拟器 → 智能体 → 经验回放 → 策略网络
↓ ↓ ↓ ↓
状态观察 → 动作选择 → 记忆存储 → 策略优化
↓ ↓ ↓ ↓
奖励反馈 → 价值估计 → 批量学习 → 梯度更新

核心组件选型

数值计算：ND4J（高效张量运算）

深度学习：Deep Java Library (DJL)

分布式计算：Apache Spark

环境模拟：自定义模拟器 + OpenAI Gym接口

可视化：JFreeChart + JavaFX

项目依赖配置

xml

1.0.0-M2.1
0.25.0
3.4.0
3.2.0

org.springframework.boot
spring-boot-starter-web
${spring-boot.version}

<!-- 数值计算 -->
<dependency>
    <groupId>org.nd4j</groupId>
    <artifactId>nd4j-api</artifactId>
    <version>${nd4j.version}</version>
</dependency>

<dependency>
    <groupId>org.nd4j</groupId>
    <artifactId>nd4j-native-platform</artifactId>
    <version>${nd4j.version}</version>
</dependency>

<!-- 深度学习 -->
<dependency>
    <groupId>ai.djl</groupId>
    <artifactId>api</artifactId>
    <version>${djl.version}</version>
</dependency>

<dependency>
    <groupId>ai.djl.pytorch</groupId>
    <artifactId>pytorch-engine</artifactId>
    <version>${djl.version}</version>
</dependency>

<!-- 分布式计算 -->
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.13</artifactId>
    <version>${spark.version}</version>
</dependency>

<!-- 可视化 -->
<dependency>
    <groupId>org.jfree</groupId>
    <artifactId>jfreechart</artifactId>
    <version>1.5.4</version>
</dependency>

<!-- 配置管理 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-configuration-processor</artifactId>
    <version>${spring-boot.version}</version>
</dependency>

三、强化学习核心组件实现

环境接口与模拟器

java
// Environment.java - 通用环境接口
public interface Environment {

/**
 * 重置环境到初始状态
 */
S reset();

/**
 * 执行动作并返回结果
 */
StepResult<S> step(A action);

/**
 * 获取当前状态
 */
S getCurrentState();

/**
 * 检查是否达到终止状态
 */
boolean isTerminal();

/**
 * 获取动作空间
 */
ActionSpace<A> getActionSpace();

/**
 * 获取状态空间
 */
StateSpace<S> getStateSpace();

/**
 * 渲染当前环境状态
 */
void render();

}

// 步进结果
@Data
@AllArgsConstructor
public class StepResult {
private S nextState;
private double reward;
private boolean done;
private Map info;
}

// 动作空间接口
public interface ActionSpace {
List getAvailableActions();
boolean contains(A action);
int getDimension();
A sample();
}

// 状态空间接口
public interface StateSpace {
int getDimension();
S getMinValues();
S getMaxValues();
}

// 网格世界环境实现
@Component
@Slf4j
public class GridWorldEnvironment implements Environment {

private final int gridSize;
private final double[][] rewards;
private final boolean[][] obstacles;
private int[] currentState;
private final int[] goalState;
private final Random random;

public GridWorldEnvironment(int gridSize) {
    this.gridSize = gridSize;
    this.rewards = new double[gridSize][gridSize];
    this.obstacles = new boolean[gridSize][gridSize];
    this.random = new Random();
    this.goalState = new int[]{gridSize - 1, gridSize - 1};

    initializeEnvironment();
    reset();
}

private void initializeEnvironment() {
    // 设置目标位置的高奖励
    rewards[goalState[0]][goalState[1]] = 10.0;

    // 随机设置障碍物
    for (int i = 0; i < gridSize * gridSize / 5; i++) {
        int x = random.nextInt(gridSize);
        int y = random.nextInt(gridSize);
        if (x != 0 || y != 0) { // 起点不能有障碍物
            obstacles[x][y] = true;
            rewards[x][y] = -5.0; // 障碍物惩罚
        }
    }

    // 设置一些负奖励区域
    for (int i = 0; i < gridSize * gridSize / 10; i++) {
        int x = random.nextInt(gridSize);
        int y = random.nextInt(gridSize);
        if (!obstacles[x][y] && (x != goalState[0] || y != goalState[1])) {
            rewards[x][y] = -2.0;
        }
    }
}

@Override
public int[] reset() {
    currentState = new int[]{0, 0}; // 从左上角开始
    return currentState.clone();
}

@Override
public StepResult<int[]> step(Integer action) {
    int[] nextState = currentState.clone();

    // 0:上, 1:右, 2:下, 3:左
    switch (action) {
        case 0: nextState[0] = Math.max(0, nextState[0] - 1); break;
        case 1: nextState[1] = Math.min(gridSize - 1, nextState[1] + 1); break;
        case 2: nextState[0] = Math.min(gridSize - 1, nextState[0] + 1); break;
        case 3: nextState[1] = Math.max(0, nextState[1] - 1); break;
    }

    // 检查障碍物
    if (obstacles[nextState[0]][nextState[1]]) {
        nextState = currentState; // 撞到障碍物，保持原位置
    }

    double reward = rewards[nextState[0]][nextState[1]] - 0.1; // 每一步的小惩罚

    boolean done = (nextState[0] == goalState[0] && nextState[1] == goalState[1]) ||
                  isTerminalState(nextState);

    currentState = nextState;

    return new StepResult<>(
        currentState.clone(),
        reward,
        done,
        Map.of("action", action, "position", currentState.clone())
    );
}

private boolean isTerminalState(int[] state) {
    return state[0] == goalState[0] && state[1] == goalState[1];
}

@Override
public int[] getCurrentState() {
    return currentState.clone();
}

@Override
public boolean isTerminal() {
    return isTerminalState(currentState);
}

@Override
public ActionSpace<Integer> getActionSpace() {
    return new DiscreteActionSpace(4); // 4个方向
}

@Override
public StateSpace<int[]> getStateSpace() {
    return new GridStateSpace(gridSize);
}

@Override
public void render() {
    for (int i = 0; i < gridSize; i++) {
        for (int j = 0; j < gridSize; j++) {
            if (i == currentState[0] && j == currentState[1]) {
                System.out.print("A "); // 智能体
            } else if (i == goalState[0] && j == goalState[1]) {
                System.out.print("G "); // 目标
            } else if (obstacles[i][j]) {
                System.out.print("X "); // 障碍物
            } else {
                System.out.print(". "); // 空地
            }
        }
        System.out.println();
    }
    System.out.println();
}

// 离散动作空间实现
private static class DiscreteActionSpace implements ActionSpace<Integer> {
    private final int size;

    public DiscreteActionSpace(int size) {
        this.size = size;
    }

    @Override
    public List<Integer> getAvailableActions() {
        List<Integer> actions = new ArrayList<>();
        for (int i = 0; i < size; i++) {
            actions.add(i);
        }
        return actions;
    }

    @Override
    public boolean contains(Integer action) {
        return action >= 0 && action < size;
    }

    @Override
    public int getDimension() {
        return size;
    }

    @Override
    public Integer sample() {
        return ThreadLocalRandom.current().nextInt(size);
    }
}

// 网格状态空间实现
private static class GridStateSpace implements StateSpace<int[]> {
    private final int gridSize;

    public GridStateSpace(int gridSize) {
        this.gridSize = gridSize;
    }

    @Override
    public int getDimension() {
        return 2; // x,y坐标
    }

    @Override
    public int[] getMinValues() {
        return new int[]{0, 0};
    }

    @Override
    public int[] getMaxValues() {
        return new int[]{gridSize - 1, gridSize - 1};
    }
}

}

~~经验回放缓冲区~~

java
// ExperienceReplayBuffer.java
@Component
@Slf4j
public class ExperienceReplayBuffer {

private final int capacity;
private final Deque<Experience<S, A>> buffer;
private final Random random;
private final PriorityQueue<PriorityExperience<S, A>> priorityBuffer;
private final double alpha; // 优先级指数
private final double beta; // 重要性采样权重
private final double epsilon; // 小常数避免零概率

public ExperienceReplayBuffer(int capacity) {
    this(capacity, 0.6, 0.4, 1e-6);
}

public ExperienceReplayBuffer(int capacity, double alpha, double beta, double epsilon) {
    this.capacity = capacity;
    this.buffer = new ArrayDeque<>(capacity);
    this.random = new Random();
    this.priorityBuffer = new PriorityQueue<>(capacity, 
        Comparator.comparingDouble(PriorityExperience::getPriority).reversed());
    this.alpha = alpha;
    this.beta = beta;
    this.epsilon = epsilon;
}

/**
 * 添加经验到缓冲区
 */
public void add(Experience<S, A> experience) {
    add(experience, getMaxPriority());
}

/**
 * 添加带优先级的经验
 */
public void add(Experience<S, A> experience, double priority) {
    PriorityExperience<S, A> priorityExp = new PriorityExperience<>(experience, priority);

    if (priorityBuffer.size() >= capacity) {
        // 移除优先级最低的经验
        priorityBuffer.poll();
    }

    priorityBuffer.offer(priorityExp);

    // 同时维护普通缓冲区用于随机采样
    if (buffer.size() >= capacity) {
        buffer.removeFirst();
    }
    buffer.addLast(experience);
}

/**
 * 从缓冲区采样一批经验
 */
public Batch<Experience<S, A>> sample(int batchSize) {
    return sample(batchSize, true);
}

/**
 * 采样经验（支持优先级采样）
 */
public Batch<Experience<S, A>> sample(int batchSize, boolean usePriority) {
    if (size() < batchSize) {
        throw new IllegalStateException("缓冲区中的经验不足");
    }

    if (!usePriority) {
        // 均匀采样
        List<Experience<S, A>> samples = new ArrayList<>();
        List<Experience<S, A>> tempBuffer = new ArrayList<>(buffer);

        for (int i = 0; i < batchSize; i++) {
            samples.add(tempBuffer.get(random.nextInt(tempBuffer.size())));
        }

        return new Batch<>(samples, null);
    } else {
        // 基于优先级的采样
        return samplePrioritized(batchSize);
    }
}

/**
 * 优先级经验采样
 */
private Batch<Experience<S, A>> samplePrioritized(int batchSize) {
    List<PriorityExperience<S, A>> prioritySamples = new ArrayList<>();
    List<Experience<S, A>> samples = new ArrayList<>();
    double[] weights = new double[batchSize];

    // 计算总优先级
    double totalPriority = priorityBuffer.stream()
            .mapToDouble(PriorityExperience::getPriority)
            .sum();

    // 基于优先级采样
    List<PriorityExperience<S, A>> tempList = new ArrayList<>(priorityBuffer);
    for (int i = 0; i < batchSize; i++) {
        double rand = random.nextDouble() * totalPriority;
        double cumulative = 0.0;

        for (PriorityExperience<S, A> exp : tempList) {
            cumulative += exp.getPriority();
            if (cumulative >= rand) {
                prioritySamples.add(exp);
                samples.add(exp.getExperience());
                break;
            }
        }
    }

    // 计算重要性采样权重
    for (int i = 0; i < batchSize; i++) {
        PriorityExperience<S, A> exp = prioritySamples.get(i);
        double probability = exp.getPriority() / totalPriority;
        weights[i] = Math.pow(probability * size(), -beta);
    }

    // 归一化权重
    double maxWeight = Arrays.stream(weights).max().orElse(1.0);
    for (int i = 0; i < weights.length; i++) {
        weights[i] /= maxWeight;
    }

    return new Batch<>(samples, weights);
}

/**
 * 更新经验的优先级（基于TD误差）
 */
public void updatePriorities(List<Experience<S, A>> experiences, List<Double> errors) {
    for (int i = 0; i < experiences.size(); i++) {
        Experience<S, A> exp = experiences.get(i);
        double error = errors.get(i);
        double priority = Math.pow(Math.abs(error) + epsilon, alpha);

        // 找到并更新对应的优先级经验
        priorityBuffer.stream()
            .filter(pe -> pe.getExperience().equals(exp))
            .findFirst()
            .ifPresent(pe -> pe.setPriority(priority));
    }
}

public int size() {
    return Math.max(buffer.size(), priorityBuffer.size());
}

public boolean isFull() {
    return size() >= capacity;
}

public void clear() {
    buffer.clear();
    priorityBuffer.clear();
}

// 经验数据类
@Data
@AllArgsConstructor
public static class Experience<S, A> {
    private S state;
    private A action;
    private double reward;
    private S nextState;
    private boolean done;

    public INDArray toINDArray(StateProcessor<S> processor) {
        // 将经验转换为NDArray用于神经网络训练
        INDArray stateArray = processor.process(state);
        INDArray nextStateArray = processor.process(nextState);

        // 这里简化实现，实际中需要更复杂的转换
        return Nd4j.concat(0, stateArray, nextStateArray);
    }
}

// 带优先级的经验
@Data
@AllArgsConstructor
private static class PriorityExperience<S, A> {
    private Experience<S, A> experience;
    private double priority;
}

// 批次数据
@Data
@AllArgsConstructor
public static class Batch<S, A> {
    private List<Experience<S, A>> experiences;
    private double[] weights; // 重要性采样权重
}

}

// 状态处理器接口
public interface StateProcessor {
INDArray process(S state);
int getOutputDimension();
}
四、核心强化学习算法实现

~~Q学习算法~~

java
// QLearningAgent.java
@Component
@Slf4j
public class QLearningAgent implements LearningAgent {

private final Environment<S, A> environment;
private final QTable<S, A> qTable;
private final double learningRate;
private final double discountFactor;
private final EpsilonGreedyStrategy<A> explorationStrategy;
private final TrainingMetrics metrics;

public QLearningAgent(Environment<S, A> environment, 
                     double learningRate, 
                     double discountFactor,
                     double initialEpsilon,
                     double epsilonDecay) {
    this.environment = environment;
    this.learningRate = learningRate;
    this.discountFactor = discountFactor;
    this.qTable = new QTable<>();
    this.explorationStrategy = new EpsilonGreedyStrategy<>(
        initialEpsilon, epsilonDecay, environment.getActionSpace());
    this.metrics = new TrainingMetrics();
}

@Override
public A chooseAction(S state) {
    return explorationStrategy.chooseAction(
        action -> qTable.getQValue(state, action),
        environment.getActionSpace().getAvailableActions()
    );
}

@Override
public void learn(ExperienceReplayBuffer.Experience<S, A> experience) {
    S state = experience.getState();
    A action = experience.getAction();
    double reward = experience.getReward();
    S nextState = experience.getNextState();
    boolean done = experience.isDone();

    double currentQ = qTable.getQValue(state, action);
    double nextMaxQ = done ? 0 : getMaxQValue(nextState);

    // Q学习更新公式
    double newQ = currentQ + learningRate * 
                 (reward + discountFactor * nextMaxQ - currentQ);

    qTable.updateQValue(state, action, newQ);

    // 记录学习指标
    metrics.recordStep(reward, newQ - currentQ, explorationStrategy.getCurrentEpsilon());
}

@Override
public void train(int episodes, int maxStepsPerEpisode) {
    log.info("开始Q学习训练，共{}回合，每回合最多{}步", episodes, maxStepsPerEpisode);

    for (int episode = 0; episode < episodes; episode++) {
        S state = environment.reset();
        double episodeReward = 0;
        int steps = 0;

        while (steps < maxStepsPerEpisode && !environment.isTerminal()) {
            // 选择并执行动作
            A action = chooseAction(state);
            Environment.StepResult<S> result = environment.step(action);

            // 学习
            ExperienceReplayBuffer.Experience<S, A> experience = 
                new ExperienceReplayBuffer.Experience<>(
                    state, action, result.getReward(), 
                    result.getNextState(), result.isDone()
                );
            learn(experience);

            state = result.getNextState();
            episodeReward += result.getReward();
            steps++;

            if (result.isDone()) {
                break;
            }
        }

        // 更新探索率
        explorationStrategy.decayEpsilon();

        // 记录回合指标
        metrics.recordEpisode(episodeReward, steps, episode);

        if (episode % 100 == 0) {
            log.info("回合 {}: 总奖励={}, 步数={}, 探索率={:.3f}", 
                    episode, episodeReward, steps, explorationStrategy.getCurrentEpsilon());
        }
    }

    log.info("训练完成");
    metrics.printSummary();
}

@Override
public double evaluate(int episodes, int maxStepsPerEpisode) {
    double totalReward = 0;

    for (int episode = 0; episode < episodes; episode++) {
        S state = environment.reset();
        double episodeReward = 0;
        int steps = 0;

        while (steps < maxStepsPerEpisode && !environment.isTerminal()) {
            // 在评估时使用贪婪策略
            A action = getBestAction(state);
            Environment.StepResult<S> result = environment.step(action);

            state = result.getNextState();
            episodeReward += result.getReward();
            steps++;

            if (result.isDone()) {
                break;
            }
        }

        totalReward += episodeReward;
    }

    double averageReward = totalReward / episodes;
    log.info("评估完成: 平均奖励={}", averageReward);
    return averageReward;
}

@Override
public A getBestAction(S state) {
    List<A> availableActions = environment.getActionSpace().getAvailableActions();
    return availableActions.stream()
            .max(Comparator.comparingDouble(action -> qTable.getQValue(state, action)))
            .orElse(environment.getActionSpace().sample());
}

private double getMaxQValue(S state) {
    List<A> availableActions = environment.getActionSpace().getAvailableActions();
    return availableActions.stream()
            .mapToDouble(action -> qTable.getQValue(state, action))
            .max()
            .orElse(0.0);
}

public QTable<S, A> getQTable() {
    return qTable;
}

public TrainingMetrics getMetrics() {
    return metrics;
}

// Q表实现
@Slf4j
public static class QTable<S, A> {
    private final Map<S, Map<A, Double>> table;
    private final double initialValue;

    public QTable() {
        this(0.0);
    }

    public QTable(double initialValue) {
        this.table = new HashMap<>();
        this.initialValue = initialValue;
    }

    public double getQValue(S state, A action) {
        return table.computeIfAbsent(state, k -> new HashMap<>())
                   .getOrDefault(action, initialValue);
    }

    public void updateQValue(S state, A action, double value) {
        table.computeIfAbsent(state, k -> new HashMap<>())
             .put(action, value);
    }

    public Map<A, Double> getActionValues(S state) {
        return table.getOrDefault(state, new HashMap<>());
    }

    public void saveToFile(String filePath) {
        try (ObjectOutputStream oos = new ObjectOutputStream(
                new FileOutputStream(filePath))) {
            oos.writeObject(table);
            log.info("Q表已保存到: {}", filePath);
        } catch (IOException e) {
            log.error("保存Q表失败", e);
        }
    }

    @SuppressWarnings("unchecked")
    public void loadFromFile(String filePath) {
        try (ObjectInputStream ois = new ObjectInputStream(
                new FileInputStream(filePath))) {
            table.clear();
            table.putAll((Map<S, Map<A, Double>>) ois.readObject());
            log.info("Q表已从 {} 加载", filePath);
        } catch (IOException | ClassNotFoundException e) {
            log.error("加载Q表失败", e);
        }
    }
}

}

~~深度Q网络（DQN）实现~~

java
// DeepQNetworkAgent.java
@Component
@Slf4j
public class DeepQNetworkAgent implements LearningAgent {

private final Environment<S, A> environment;
private final StateProcessor<S> stateProcessor;
private final Model qNetwork;
private final Model targetNetwork;
private final ExperienceReplayBuffer<S, A> replayBuffer;
private final EpsilonGreedyStrategy<A> explorationStrategy;
private final TrainingConfig config;
private final TrainingMetrics metrics;
private final NDManager manager;

private int trainStep = 0;

public DeepQNetworkAgent(Environment<S, A> environment,
                       StateProcessor<S> stateProcessor,
                       Model qNetwork,
                       ExperienceReplayBuffer<S, A> replayBuffer,
                       TrainingConfig config) {
    this.environment = environment;
    this.stateProcessor = stateProcessor;
    this.qNetwork = qNetwork;
    this.targetNetwork = copyModel(qNetwork); // 目标网络
    this.replayBuffer = replayBuffer;
    this.config = config;
    this.manager = NDManager.newBaseManager();

    this.explorationStrategy = new EpsilonGreedyStrategy<>(
        config.getInitialEpsilon(),
        config.getEpsilonDecay(),
        environment.getActionSpace()
    );

    this.metrics = new TrainingMetrics();
}

@Override
public A chooseAction(S state) {
    // 使用ε-贪婪策略选择动作
    if (Math.random() < explorationStrategy.getCurrentEpsilon()) {
        return environment.getActionSpace().sample(); // 探索
    } else {
        return getBestAction(state); // 利用
    }
}

@Override
public A getBestAction(S state) {
    try {
        INDArray stateTensor = stateProcessor.process(state);

        // 使用Q网络预测Q值
        try (Predictor predictor = qNetwork.newPredictor()) {
            NDList input = new NDList(stateTensor);
            NDList output = predictor.predict(input);
            INDArray qValues = output.get(0);

            // 选择最大Q值对应的动作
            int bestAction = Nd4j.argMax(qValues, 1).getInt(0);
            return environment.getActionSpace().getAvailableActions().get(bestAction);
        }
    } catch (Exception e) {
        log.error("选择最佳动作失败", e);
        return environment.getActionSpace().sample();
    }
}

@Override
public void learn(ExperienceReplayBuffer.Experience<S, A> experience) {
    // 深度Q学习使用经验回放，不在单个经验上立即学习
    replayBuffer.add(experience);

    // 定期从回放缓冲区采样并训练
    if (replayBuffer.size() >= config.getBatchSize()) {
        trainFromReplayBuffer();
    }
}

@Override
public void train(int episodes, int maxStepsPerEpisode) {
    log.info("开始深度Q网络训练，共{}回合", episodes);

    for (int episode = 0; episode < episodes; episode++) {
        S state = environment.reset();
        double episodeReward = 0;
        int steps = 0;

        while (steps < maxStepsPerEpisode && !environment.isTerminal()) {
            // 选择并执行动作
            A action = chooseAction(state);
            Environment.StepResult<S> result = environment.step(action);

            // 存储经验
            ExperienceReplayBuffer.Experience<S, A> experience = 
                new ExperienceReplayBuffer.Experience<>(
                    state, action, result.getReward(), 
                    result.getNextState(), result.isDone()
                );
            replayBuffer.add(experience);

            state = result.getNextState();
            episodeReward += result.getReward();
            steps++;
            trainStep++;

            // 定期训练
            if (replayBuffer.size() >= config.getBatchSize() && 
                trainStep % config.getTrainFrequency() == 0) {
                trainFromReplayBuffer();
            }

            // 定期更新目标网络
            if (trainStep % config.getTargetUpdateFrequency() == 0) {
                updateTargetNetwork();
            }

            if (result.isDone()) {
                break;
            }
        }

        // 更新探索率
        explorationStrategy.decayEpsilon();

        // 记录指标
        metrics.recordEpisode(episodeReward, steps, episode);

        if (episode % config.getLogFrequency() == 0) {
            double avgQ = computeAverageQValue();
            log.info("回合 {}: 奖励={}, 步数={}, 探索率={:.3f}, 平均Q值={:.3f}", 
                    episode, episodeReward, steps, 
                    explorationStrategy.getCurrentEpsilon(), avgQ);
        }

        // 定期评估
        if (episode % config.getEvalFrequency() == 0) {
            double evalReward = evaluate(5, maxStepsPerEpisode);
            metrics.recordEvalEpisode(evalReward, episode);
        }
    }

    log.info("深度Q网络训练完成");
    metrics.printSummary();
}

/**
 * 从经验回放缓冲区训练
 */
private void trainFromReplayBuffer() {
    try {
        // 采样一批经验
        ExperienceReplayBuffer.Batch<S, A> batch = 
            replayBuffer.sample(config.getBatchSize(), config.isUsePriority());

        List<INDArray> stateBatch = new ArrayList<>();
        List<INDArray> nextStateBatch = new ArrayList<>();
        List<Integer> actionBatch = new ArrayList<>();
        List<Double> rewardBatch = new ArrayList<>();
        List<Boolean> doneBatch = new ArrayList<>();

        // 准备训练数据
        for (ExperienceReplayBuffer.Experience<S, A> exp : batch.getExperiences()) {
            stateBatch.add(stateProcessor.process(exp.getState()));
            nextStateBatch.add(stateProcessor.process(exp.getNextState()));
            actionBatch.add(getActionIndex(exp.getAction()));
            rewardBatch.add(exp.getReward());
            doneBatch.add(exp.isDone());
        }

        // 转换为NDArray
        INDArray states = Nd4j.concat(0, stateBatch.toArray(new INDArray[0]));
        INDArray nextStates = Nd4j.concat(0, nextStateBatch.toArray(new INDArray[0]));
        INDArray actions = Nd4j.create(actionBatch.stream()
                .mapToDouble(Integer::doubleValue).toArray(), new long[]{config.getBatchSize(), 1});
        INDArray rewards = Nd4j.create(rewardBatch.stream()
                .mapToDouble(Double::doubleValue).toArray(), new long[]{config.getBatchSize(), 1});
        INDArray dones = Nd4j.create(doneBatch.stream()
                .mapToDouble(b -> b ? 1.0 : 0.0).toArray(), new long[]{config.getBatchSize(), 1});

        // 计算目标Q值
        INDArray targetQValues = computeTargetQValues(nextStates, rewards, dones);

        // 训练Q网络
        trainQNetwork(states, actions, targetQValues, batch.getWeights());

    } catch (Exception e) {
        log.error("从回放缓冲区训练失败", e);
    }
}

/**
 * 计算目标Q值
 */
private INDArray computeTargetQValues(INDArray nextStates, INDArray rewards, INDArray dones) {
    try (Predictor predictor = targetNetwork.newPredictor()) {
        // 使用目标网络计算下一个状态的Q值
        NDList nextOutput = predictor.predict(new NDList(nextStates));
        INDArray nextQValues = nextOutput.get(0);

        // 选择最大Q值
        INDArray maxNextQ = nextQValues.max(1);

        // 计算目标Q值: r + γ * maxQ * (1 - done)
        INDArray targetQ = rewards.add(
            maxNextQ.mul(config.getDiscountFactor()).mul(dones.rsub(1.0))
        );

        return targetQ.reshape(config.getBatchSize(), 1);
    } catch (Exception e) {
        log.error("计算目标Q值失败", e);
        return rewards; // 失败时返回即时奖励
    }
}

/**
 * 训练Q网络
 */
private void trainQNetwork(INDArray states, INDArray actions, 
                         INDArray targetQValues, double[] weights) {
    try (Trainer trainer = qNetwork.newTrainer(config.getTrainingConfig())) {
        // 准备训练数据
        NDList inputs = new NDList(states, actions);
        NDList labels = new NDList(targetQValues);

        // 执行训练步骤
        EasyTrain.trainBatch(trainer, inputs, labels);
        trainer.step();

        // 记录损失
        double loss = trainer.getTrainingResult().getEvaluations().get("loss");
        metrics.recordLoss(loss);

        // 如果使用优先级回放，更新经验的优先级
        if (config.isUsePriority() && weights != null) {
            updateExperiencePriorities(states, actions, targetQValues, weights);
        }

    } catch (Exception e) {
        log.error("训练Q网络失败", e);
    }
}

/**
 * 更新经验优先级（基于TD误差）
 */
private void updateExperiencePriorities(INDArray states, INDArray actions, 
                                      INDArray targetQValues, double[] weights) {
    try (Predictor predictor = qNetwork.newPredictor()) {
        // 计算当前Q值
        NDList currentOutput = predictor.predict(new NDList(states));
        INDArray currentQValues = currentOutput.get(0);

        // 计算TD误差
        List<Double> errors = new ArrayList<>();
        for (int i = 0; i < config.getBatchSize(); i++) {
            int action = actions.getInt(i, 0);
            double currentQ = currentQValues.getDouble(i, action);
            double targetQ = targetQValues.getDouble(i, 0);
            double error = Math.abs(targetQ - currentQ);
            errors.add(error);
        }

        // 更新优先级（这里简化实现，实际需要更复杂的逻辑）
        // replayBuffer.updatePriorities(batch.getExperiences(), errors);

    } catch (Exception e) {
        log.error("更新经验优先级失败", e);
    }
}

/**
 * 更新目标网络参数
 */
private void updateTargetNetwork() {
    // 使用软更新：θ' = τ * θ + (1 - τ) * θ'
    float tau = config.getTargetUpdateTau();

    // 获取Q网络和目标网络的参数
    Map<String, Parameter> qParams = qNetwork.getBlock().getParameters();
    Map<String, Parameter> targetParams = targetNetwork.getBlock().getParameters();

    for (String paramName : qParams.keySet()) {
        Parameter qParam = qParams.get(paramName);
        Parameter targetParam = targetParams.get(paramName);

        if (qParam != null && targetParam != null) {
            try (NDArray qArray = qParam.getArray();
                 NDArray targetArray = targetParam.getArray()) {

                // 软更新
                targetArray.muli(1 - tau).addi(qArray.mul(tau));
            }
        }
    }

    log.debug("目标网络已更新 (τ={})", tau);
}

/**
 * 计算平均Q值（用于监控）
 */
private double computeAverageQValue() {
    try {
        // 从回放缓冲区采样一些状态
        ExperienceReplayBuffer.Batch<S, A> batch = 
            replayBuffer.sample(Math.min(100, replayBuffer.size()), false);

        double totalQ = 0;
        int count = 0;

        for (ExperienceReplayBuffer.Experience<S, A> exp : batch.getExperiences()) {
            INDArray state = stateProcessor.process(exp.getState());

            try (Predictor predictor = qNetwork.newPredictor()) {
                NDList output = predictor.predict(new NDList(state));
                INDArray qValues = output.get(0);
                totalQ += qValues.mean().getDouble();
                count++;
            }
        }

        return count > 0 ? totalQ / count : 0.0;

    } catch (Exception e) {
        log.error("计算平均Q值失败", e);
        return 0.0;
    }
}

@Override
public double evaluate(int episodes, int maxStepsPerEpisode) {
    double totalReward = 0;

    for (int episode = 0; episode < episodes; episode++) {
        S state = environment.reset();
        double episodeReward = 0;
        int steps = 0;

        while (steps < maxStepsPerEpisode && !environment.isTerminal()) {
            A action = getBestAction(state);
            Environment.StepResult<S> result = environment.step(action);

            state = result.getNextState();
            episodeReward += result.getReward();
            steps++;

            if (result.isDone()) {
                break;
            }
        }

        totalReward += episodeReward;
    }

    return totalReward / episodes;
}

private int getActionIndex(A action) {
    List<A> availableActions = environment.getActionSpace().getAvailableActions();
    return availableActions.indexOf(action);
}

private Model copyModel(Model original) {
    // 创建模型的深拷贝
    // 简化实现，实际需要更复杂的复制逻辑
    try {
        Block block = original.getBlock();
        Model copy = Model.newInstance("target_network");
        copy.setBlock(block);
        return copy;
    } catch (Exception e) {
        throw new RuntimeException("复制模型失败", e);
    }
}

public void saveModel(String modelPath) {
    try {
        qNetwork.save(Paths.get(modelPath), "q_network");
        log.info("Q网络模型已保存到: {}", modelPath);
    } catch (Exception e) {
        log.error("保存模型失败", e);
    }
}

public void loadModel(String modelPath) {
    try {
        qNetwork.load(Paths.get(modelPath));
        updateTargetNetwork(); // 同时更新目标网络
        log.info("Q网络模型已从 {} 加载", modelPath);
    } catch (Exception e) {
        log.error("加载模型失败", e);
    }
}

@PreDestroy
public void cleanup() {
    if (manager != null) {
        manager.close();
    }
    if (qNetwork != null) {
        qNetwork.close();
    }
    if (targetNetwork != null) {
        targetNetwork.close();
    }
}

}
五、策略梯度方法实现

~~演员-评论家算法~~

java
// ActorCriticAgent.java
@Component
@Slf4j
public class ActorCriticAgent implements LearningAgent {

private final Environment<S, A> environment;
private final StateProcessor<S> stateProcessor;
private final Model actorNetwork; // 策略网络
private final Model criticNetwork; // 价值网络
private final TrainingConfig config;
private final TrainingMetrics metrics;
private final NDManager manager;

public ActorCriticAgent(Environment<S, A> environment,
                      StateProcessor<S> stateProcessor,
                      Model actorNetwork,
                      Model criticNetwork,
                      TrainingConfig config) {
    this.environment = environment;
    this.stateProcessor = stateProcessor;
    this.actorNetwork = actorNetwork;
    this.criticNetwork = criticNetwork;
    this.config = config;
    this.manager = NDManager.newBaseManager();
    this.metrics = new TrainingMetrics();
}

@Override
public A chooseAction(S state) {
    try {
        INDArray stateTensor = stateProcessor.process(state);

        // 使用演员网络预测动作概率
        try (Predictor predictor = actorNetwork.newPredictor()) {
            NDList output = predictor.predict(new NDList(stateTensor));
            INDArray actionProbs = output.get(0);

            // 根据概率分布采样动作
            return sampleAction(actionProbs);
        }
    } catch (Exception e) {
        log.error("选择动作失败", e);
        return environment.getActionSpace().sample();
    }
}

@Override
public A getBestAction(S state) {
    try {
        INDArray stateTensor = stateProcessor.process(state);

        // 使用演员网络预测动作概率
        try (Predictor predictor = actorNetwork.newPredictor()) {
            NDList output = predictor.predict(new NDList(stateTensor));
            INDArray actionProbs = output.get(0);

            // 选择概率最高的动作
            int bestAction = Nd4j.argMax(actionProbs, 1).getInt(0);
            return environment.getActionSpace().getAvailableActions().get(bestAction);
        }
    } catch (Exception e) {
        log.error("选择最佳动作失败", e);
        return environment.getActionSpace().sample();
    }
}

@Override
public void learn(ExperienceReplayBuffer.Experience<S, A> experience) {
    // 演员-评论家算法通常使用在线学习，这里简化实现
    // 实际中会收集一个轨迹然后学习
}

@Override
public void train(int episodes, int maxStepsPerEpisode) {
    log.info("开始演员-评论家训练，共{}回合", episodes);

    for (int episode = 0; episode < episodes; episode++) {
        List<ExperienceReplayBuffer.Experience<S, A>> trajectory = 
            new ArrayList<>();

        S state = environment.reset();
        double episodeReward = 0;
        int steps = 0;

        // 收集轨迹
        while (steps < maxStepsPerEpisode && !environment.isTerminal()) {
            A action = chooseAction(state);
            Environment.StepResult<S> result = environment.step(action);

            ExperienceReplayBuffer.Experience<S, A> experience = 
                new ExperienceReplayBuffer.Experience<>(
                    state, action, result.getReward(), 
                    result.getNextState(), result.isDone()
                );
            trajectory.add(experience);

            state = result.getNextState();
            episodeReward += result.getReward();
            steps++;

            if (result.isDone()) {
                break;
            }
        }

        // 使用收集的轨迹进行学习
        learnFromTrajectory(trajectory);

        // 记录指标
        metrics.recordEpisode(episodeReward, steps, episode);

        if (episode % config.getLogFrequency() == 0) {
            log.info("回合 {}: 奖励={}, 步数={}", episode, episodeReward, steps);
        }
    }

    log.info("演员-评论家训练完成");
    metrics.printSummary();
}

/**
 * 从轨迹学习（使用优势演员-评论家）
 */
private void learnFromTrajectory(List<ExperienceReplayBuffer.Experience<S, A>> trajectory) {
    if (trajectory.isEmpty()) return;

    try {
        // 准备训练数据
        List<INDArray> states = new ArrayList<>();
        List<INDArray> actions = new ArrayList<>();
        List<Double> rewards = new ArrayList<>();
        List<INDArray> nextStates = new ArrayList<>();
        List<Boolean> dones = new ArrayList<>();

        for (ExperienceReplayBuffer.Experience<S, A> exp : trajectory) {
            states.add(stateProcessor.process(exp.getState()));
            actions.add(getActionOneHot(exp.getAction()));
            rewards.add(exp.getReward());
            nextStates.add(stateProcessor.process(exp.getNextState()));
            dones.add(exp.isDone());
        }

        // 计算折扣回报和优势函数
        double[] returns = computeReturns(rewards, config.getDiscountFactor());
        double[] advantages = computeAdvantages(states, nextStates, rewards, dones);

        // 训练演员网络（策略梯度）
        trainActorNetwork(states, actions, advantages);

        // 训练评论家网络（价值函数拟合）
        trainCriticNetwork(states, returns);

    } catch (Exception e) {
        log.error("从轨迹学习失败", e);
    }
}

/**
 * 计算折扣回报
 */
private double[] computeReturns(List<Double> rewards, double gamma) {
    double[] returns = new double[rewards.size()];
    double cumulative = 0;

    // 从后向前计算
    for (int i = rewards.size() - 1; i >= 0; i--) {
        cumulative = rewards.get(i) + gamma * cumulative;
        returns[i] = cumulative;
    }

    return returns;
}

/**
 * 计算优势函数
 */
private double[] computeAdvantages(List<INDArray> states, List<INDArray> nextStates,
                                 List<Double> rewards, List<Boolean> dones) {
    double[] advantages = new double[states.size()];

    try (Predictor criticPredictor = criticNetwork.newPredictor()) {
        for (int i = 0; i < states.size(); i++) {
            // 计算当前状态价值
            NDList currentOutput = criticPredictor.predict(new NDList(states.get(i)));
            double currentValue = currentOutput.get(0).getDouble();

            // 计算下一个状态价值
            double nextValue = 0;
            if (!dones.get(i)) {
                NDList nextOutput = criticPredictor.predict(new NDList(nextStates.get(i)));
                nextValue = nextOutput.get(0).getDouble();
            }

            // 优势函数: A = r + γV(s') - V(s)
            advantages[i] = rewards.get(i) + config.getDiscountFactor() * nextValue - currentValue;
        }
    } catch (Exception e) {
        log.error("计算优势函数失败", e);
        Arrays.fill(advantages, 0.0);
    }

    return advantages;
}

/**
 * 训练演员网络
 */
private void trainActorNetwork(List<INDArray> states, List<INDArray> actions, 
                             double[] advantages) {
    try (Trainer trainer = actorNetwork.newTrainer(config.getTrainingConfig())) {
        for (int i = 0; i < states.size(); i++) {
            // 策略梯度更新
            NDList inputs = new NDList(states.get(i), actions.get(i));
            NDList labels = new NDList(Nd4j.scalar(advantages[i]));

            EasyTrain.trainBatch(trainer, inputs, labels);
            trainer.step();
        }

        double loss = trainer.getTrainingResult().getEvaluations().get("loss");
        metrics.recordLoss(loss);

    } catch (Exception e) {
        log.error("训练演员网络失败", e);
    }
}

/**
 * 训练评论家网络
 */
private void trainCriticNetwork(List<INDArray> states, double[] returns) {
    try (Trainer trainer = criticNetwork.newTrainer(config.getTrainingConfig())) {
        for (int i = 0; i < states.size(); i++) {
            // 价值函数回归
            NDList inputs = new NDList(states.get(i));
            NDList labels = new NDList(Nd4j.scalar(returns[i]));

            EasyTrain.trainBatch(trainer, inputs, labels);
            trainer.step();
        }

        double loss = trainer.getTrainingResult().getEvaluations().get("loss");
        metrics.recordLoss(loss);

    } catch (Exception e) {
        log.error("训练评论家网络失败", e);
    }
}

/**
 * 根据概率分布采样动作
 */
private A sampleAction(INDArray actionProbs) {
    double[] probs = actionProbs.toDoubleVector();
    double random = Math.random();
    double cumulative = 0.0;

    List<A> availableActions = environment.getActionSpace().getAvailableActions();

    for (int i = 0; i < probs.length; i++) {
        cumulative += probs[i];
        if (random <= cumulative) {
            return availableActions.get(i);
        }
    }

    return availableActions.get(availableActions.size() - 1);
}

/**
 * 将动作转换为one-hot编码
 */
private INDArray getActionOneHot(A action) {
    List<A> availableActions = environment.getActionSpace().getAvailableActions();
    int actionIndex = availableActions.indexOf(action);
    int actionSize = availableActions.size();

    INDArray oneHot = Nd4j.zeros(1, actionSize);
    oneHot.putScalar(0, actionIndex, 1.0);
    return oneHot;
}

@Override
public double evaluate(int episodes, int maxStepsPerEpisode) {
    double totalReward = 0;

    for (int episode = 0; episode < episodes; episode++) {
        S state = environment.reset();
        double episodeReward = 0;
        int steps = 0;

        while (steps < maxStepsPerEpisode && !environment.isTerminal()) {
            A action = getBestAction(state);
            Environment.StepResult<S> result = environment.step(action);

            state = result.getNextState();
            episodeReward += result.getReward();
            steps++;

            if (result.isDone()) {
                break;
            }
        }

        totalReward += episodeReward;
    }

    return totalReward / episodes;
}

public void saveModels(String actorPath, String criticPath) {
    try {
        actorNetwork.save(Paths.get(actorPath), "actor_network");
        criticNetwork.save(Paths.get(criticPath), "critic_network");
        log.info("演员-评论家模型已保存");
    } catch (Exception e) {
        log.error("保存模型失败", e);
    }
}

@PreDestroy
public void cleanup() {
    if (manager != null) {
        manager.close();
    }
    if (actorNetwork != null) {
        actorNetwork.close();
    }
    if (criticNetwork != null) {
        criticNetwork.close();
    }
}

}
六、训练配置与监控系统

~~统一训练配置~~

java
// TrainingConfig.java
@Data
@ConfigurationProperties(prefix = "rl.training")
public class TrainingConfig {

// 基本参数
private double learningRate = 0.001;
private double discountFactor = 0.99;
private int batchSize = 32;
private int replayBufferSize = 10000;

// 探索参数
private double initialEpsilon = 1.0;
private double epsilonDecay = 0.995;
private double minEpsilon = 0.01;

// 训练参数
private int trainFrequency = 4;
private int targetUpdateFrequency = 100;
private float targetUpdateTau = 0.01f;

// 日志参数
private int logFrequency = 100;
private int evalFrequency = 500;

// 优先级回放
private boolean usePriority = true;
private double priorityAlpha = 0.6;
private double priorityBeta = 0.4;
private double priorityEpsilon = 1e-6;

// 网络架构
private List<Integer> hiddenLayers = List.of(128, 64);
private String activation = "relu";

public TrainingConfig.TrainingListener getTrainingListener() {
    return new DefaultTrainingListener();
}

}

// 训练指标监控
@Component
@Slf4j
public class TrainingMetrics {

private final List<Double> episodeRewards;
private final List<Double> episodeLengths;
private final List<Double> losses;
private final List<Double> evalRewards;
private final Map<String, Object> summaryStats;

public TrainingMetrics() {
    this.episodeRewards = new ArrayList<>();
    this.episodeLengths = new ArrayList<>();
    this.losses = new ArrayList<>();
    this.evalRewards = new ArrayList<>();
    this.summaryStats = new HashMap<>();
}

public void recordEpisode(double reward, int steps, int episode) {
    episodeRewards.add(reward);
    episodeLengths.add((double) steps);

    // 更新滑动窗口统计
    updateMovingAverages();
}

public void recordStep(double reward, double tdError, double epsilon) {
    // 记录单步信息
    summaryStats.put("last_reward", reward);
    summaryStats.put("last_td_error", tdError);
    summaryStats.put("current_epsilon", epsilon);
}

public void recordLoss(double loss) {
    losses.add(loss);
}

public void recordEvalEpisode(double reward, int episode) {
    evalRewards.add(reward);
    summaryStats.put("last_eval_reward", reward);
    summaryStats.put("last_eval_episode", episode);
}

public void printSummary() {
    if (episodeRewards.isEmpty()) return;

    double avgReward = episodeRewards.stream().mapToDouble(Double::doubleValue).average().orElse(0);
    double avgLength = episodeLengths.stream().mapToDouble(Double::doubleValue).average().orElse(0);
    double avgLoss = losses.stream().mapToDouble(Double::doubleValue).average().orElse(0);

    log.info("训练总结:");
    log.info("平均回合奖励: {:.3f}", avgReward);
    log.info("平均回合长度: {:.1f}", avgLength);
    log.info("平均损失: {:.5f}", avgLoss);

    if (!evalRewards.isEmpty()) {
        double avgEvalReward = evalRewards.stream().mapToDouble(Double::doubleValue).average().orElse(0);
        log.info("平均评估奖励: {:.3f}", avgEvalReward);
    }
}

private void updateMovingAverages() {
    if (episodeRewards.size() >= 10) {
        double recentAvgReward = episodeRewards.stream()
                .skip(episodeRewards.size() - 10)
                .mapToDouble(Double::doubleValue)
                .average()
                .orElse(0);
        summaryStats.put("recent_avg_reward", recentAvgReward);
    }
}

public Map<String, Object> getCurrentStats() {
    Map<String, Object> stats = new HashMap<>(summaryStats);
    stats.put("total_episodes", episodeRewards.size());
    stats.put("total_steps", episodeLengths.stream().mapToDouble(Double::doubleValue).sum());
    return stats;
}

// 可视化数据获取
public List<Double> getRewardHistory() {
    return new ArrayList<>(episodeRewards);
}

public List<Double> getLossHistory() {
    return new ArrayList<>(losses);
}

public List<Double> getEvalHistory() {
    return new ArrayList<>(evalRewards);
}

}
七、应用场景与REST API

~~强化学习服务控制器~~

java
// RLServiceController.java
@RestController
@RequestMapping("/api/rl")
@Slf4j
public class RLServiceController {

private final Map<String, LearningAgent<?, ?>> agents;
private final Map<String, Environment<?, ?>> environments;
private final TrainingConfig trainingConfig;

public RLServiceController(TrainingConfig trainingConfig,
                         List<LearningAgent<?, ?>> agentList,
                         List<Environment<?, ?>> environmentList) {
    this.trainingConfig = trainingConfig;
    this.agents = new ConcurrentHashMap<>();
    this.environments = new ConcurrentHashMap<>();

    // 注册环境和智能体
    for (Environment<?, ?> env : environmentList) {
        environments.put(env.getClass().getSimpleName(), env);
    }

    for (LearningAgent<?, ?> agent : agentList) {
        agents.put(agent.getClass().getSimpleName(), agent);
    }
}

@PostMapping("/train/{agentName}/{envName}")
public ResponseEntity<TrainingResponse> startTraining(
        @PathVariable String agentName,
        @PathVariable String envName,
        @RequestBody TrainingRequest request) {

    try {
        LearningAgent<?, ?> agent = agents.get(agentName);
        Environment<?, ?> environment = environments.get(envName);

        if (agent == null || environment == null) {
            return ResponseEntity.badRequest().body(
                TrainingResponse.error("智能体或环境不存在"));
        }

        // 异步执行训练
        CompletableFuture.runAsync(() -> {
            try {
                agent.train(request.getEpisodes(), request.getMaxSteps());
            } catch (Exception e) {
                log.error("训练失败", e);
            }
        });

        return ResponseEntity.ok(TrainingResponse.started(
            agentName, envName, request.getEpisodes()));

    } catch (Exception e) {
        log.error("启动训练失败", e);
        return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                .body(TrainingResponse.error(e.getMessage()));
    }
}

@PostMapping("/evaluate/{agentName}/{envName}")
public ResponseEntity<EvaluationResponse> evaluateAgent(
        @PathVariable String agentName,
        @PathVariable String envName,
        @RequestParam(defaultValue = "10") int episodes) {

    try {
        LearningAgent<?, ?> agent = agents.get(agentName);
        Environment<?, ?> environment = environments.get(envName);

        if (agent == null || environment == null) {
            return ResponseEntity.badRequest().body(
                EvaluationResponse.error("智能体或环境不存在"));
        }

        double averageReward = agent.evaluate(episodes, 1000);

        return ResponseEntity.ok(EvaluationResponse.success(
            agentName, envName, averageReward, episodes));

    } catch (Exception e) {
        log.error("评估失败", e);
        return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                .body(EvaluationResponse.error(e.getMessage()));
    }
}

@GetMapping("/agents")
public ResponseEntity<List<AgentInfo>> getAgents() {
    List<AgentInfo> agentInfos = agents.keySet().stream()
            .map(name -> new AgentInfo(name, "运行中"))
            .collect(Collectors.toList());
    return ResponseEntity.ok(agentInfos);
}

@GetMapping("/environments")
public ResponseEntity<List<EnvironmentInfo>> getEnvironments() {
    List<EnvironmentInfo> envInfos = environments.keySet().stream()
            .map(name -> new EnvironmentInfo(name, "可用"))
            .collect(Collectors.toList());
    return ResponseEntity.ok(envInfos);
}

@GetMapping("/metrics/{agentName}")
public ResponseEntity<TrainingMetrics> getTrainingMetrics(@PathVariable String agentName) {
    LearningAgent<?, ?> agent = agents.get(agentName);
    if (agent == null) {
        return ResponseEntity.notFound().build();
    }

    // 这里需要根据具体实现获取指标
    return ResponseEntity.ok(new TrainingMetrics());
}

// DTO类
@Data
public static class TrainingRequest {
    private int episodes = 1000;
    private int maxSteps = 1000;
    private Map<String, Object> parameters;
}

@Data
@AllArgsConstructor
public static class TrainingResponse {
    private String status;
    private String message;
    private String agentName;
    private String envName;
    private Integer episodes;

    public static TrainingResponse started(String agentName, String envName, int episodes) {
        return new TrainingResponse("started", "训练已启动", agentName, envName, episodes);
    }

    public static TrainingResponse error(String message) {
        return new TrainingResponse("error", message, null, null, null);
    }
}

@Data
@AllArgsConstructor
public static class EvaluationResponse {
    private String status;
    private String message;
    private String agentName;
    private String envName;
    private Double averageReward;
    private Integer episodes;

    public static EvaluationResponse success(String agentName, String envName, 
                                           double averageReward, int episodes) {
        return new EvaluationResponse("success", "评估完成", 
            agentName, envName, averageReward, episodes);
    }

    public static EvaluationResponse error(String message) {
        return new EvaluationResponse("error", message, null, null, null, null);
    }
}

@Data
@AllArgsConstructor
public static class AgentInfo {
    private String name;
    private String status;
}

@Data
@AllArgsConstructor
public static class EnvironmentInfo {
    private String name;
    private String status;
}

}
八、生产配置与优化

~~应用配置~~

~~yaml~~

application.yml

spring:
application:
name: java-reinforcement-learning

rl:
training:
learning-rate: 0.001
discount-factor: 0.99
batch-size: 64
replay-buffer-size: 100000

# 探索参数
initial-epsilon: 1.0
epsilon-decay: 0.995
min-epsilon: 0.01

# 网络参数
hidden-layers: [256, 128, 64]
activation: relu

# 训练调度
train-frequency: 4
target-update-frequency: 1000
target-update-tau: 0.01

# 日志和评估
log-frequency: 100
eval-frequency: 1000

logging:
level:
com.example.rl: INFO
file:
name: /var/log/rl-service.log

management:
endpoints:
web:
exposure:
include: health,metrics,info
endpoint:
health:
show-details: always

~~性能优化配置~~

java
// PerformanceOptimizer.java
@Component
@Slf4j
public class PerformanceOptimizer {

private final NDManager manager;
private final MemoryWorkspace workspace;
private final TrainingMetrics metrics;

public PerformanceOptimizer() {
    this.manager = NDManager.newBaseManager();
    this.workspace = manager.getWorkspaceManager()
            .createNewWorkspace(1024 * 1024 * 512); // 512MB工作空间
    this.metrics = new TrainingMetrics();
}

/**
 * 优化ND4J性能配置
 */
@PostConstruct
public void optimizeND4J() {
    // 设置BLAS后端
    System.setProperty("org.nd4j.linalg.default.backend", "nd4j-native");
    System.setProperty("org.nd4j.linalg.tensor.parallelism", "4");

    // 启用内存优化
    Nd4j.getMemoryManager().setAutoGcWindow(5000);
    Nd4j.getMemoryManager().togglePeriodicGc(false);

    log.info("ND4J性能优化完成");
}

/**
 * 批量推理优化
 */
public INDArray batchPredict(Model model, List<INDArray> inputs) {
    try (MemoryWorkspace ws = workspace.notifyScopeEntered()) {
        // 合并输入批次
        INDArray batchInput = Nd4j.concat(0, inputs.toArray(new INDArray[0]));

        try (Predictor predictor = model.newPredictor()) {
            NDList output = predictor.predict(new NDList(batchInput));
            return output.get(0);
        }
    } catch (Exception e) {
        log.error("批量推理失败", e);
        return null;
    }
}

/**
 * 异步训练支持
 */
public CompletableFuture<Void> trainAsync(Runnable trainingTask) {
    return CompletableFuture.runAsync(() -> {
        try (MemoryWorkspace ws = workspace.notifyScopeEntered()) {
            trainingTask.run();
        } catch (Exception e) {
            log.error("异步训练失败", e);
            throw new RuntimeException(e);
        }
    });
}

/**
 * 内存使用监控
 */
@Scheduled(fixedRate = 30000)
public void monitorMemoryUsage() {
    long usedMemory = Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory();
    long maxMemory = Runtime.getRuntime().maxMemory();

    double usagePercent = (double) usedMemory / maxMemory * 100;

    if (usagePercent > 80) {
        log.warn("内存使用率过高: {:.1f}%", usagePercent);
        // 触发垃圾回收
        System.gc();
    }

    metrics.recordMemoryUsage(usagePercent);
}

@PreDestroy
public void cleanup() {
    if (workspace != null) {
        workspace.destroyWorkspace();
    }
    if (manager != null) {
        manager.close();
    }
}

}
九、应用场景与总结

~~典型应用场景~~

~~游戏AI：训练智能体玩Atari游戏、棋类游戏等~~

~~机器人控制：机械臂控制、自动驾驶、无人机导航~~

~~资源调度：云计算资源分配、网络路由优化~~

~~推荐系统：个性化内容推荐、广告投放策略~~

~~金融交易：自动化交易策略、投资组合管理~~

~~系统优势总结~~

~~端到端学习：直接从原始输入学习最优策略~~

~~持续改进：随着经验积累不断优化性能~~

~~适应性强：能够处理动态变化的环境~~

~~泛化能力：学到的策略可以迁移到类似任务~~

~~技术挑战与解决方案~~

~~样本效率：通过经验回放、优先级采样提高数据利用率~~

~~训练稳定性：使用目标网络、梯度裁剪等技术~~

~~探索-利用平衡：ε-贪婪、熵正则化等方法~~

~~高维状态空间：使用深度神经网络进行函数逼近~~

总结

~~通过本文的实践，我们成功构建了一个完整的Java强化学习系统，具备以下核心能力：~~

~~多种算法支持：Q学习、深度Q网络、演员-评论家等~~

~~高效经验管理：优先级经验回放、批量训练~~

~~深度集成：与DJL、ND4J等深度学习框架深度集成~~

~~生产就绪：REST API、监控、配置管理~~

~~可扩展架构：支持新算法和环境的快速集成~~

强化学习代表了AI系统从被动响应到主动学习的根本性转变。Java在企业级系统中的优势与强化学习的自主学习能力相结合，为构建真正智能的自适应系统开辟了新的可能性。随着算法的不断进步和计算资源的增长，基于Java的强化学习系统将在自动化决策、智能控制等领域发挥越来越重要的作用。

Java与AI模型部署：构建企业级模型服务与生命周期管理平台

deployment.yaml

application-production.yml

application.yml

热门文章

最新文章

相关课程

相关电子书

相关实验场景

探索云世界

热门

云计算

大数据

云原生

人工智能

数据库

开发与运维

活动广场

任务中心

训练营

直播

乘风者计划

下载

镜像站

技术资料

Java与AI模型部署：构建企业级模型服务与生命周期管理平台

deployment.yaml

application-production.yml

application.yml

热门文章

最新文章

相关课程

相关电子书

相关实验场景