item3 - 9 文件解析模块2

vectorizationService.vectorize(task.getFileMd5(), task.getUserId(), task.getOrgTag(), task.isPublic());

logger.info("开始向量化文件，fileMd5: {}, userId: {}, orgTag: {}, isPublic: {}", fileMd5, userId, orgTag, isPublic);
            
// 获取文件分块内容
List<TextChunk> chunks = fetchTextChunks(fileMd5);
if (chunks == null || chunks.isEmpty()) {
    logger.warn("未找到分块内容，fileMd5: {}", fileMd5);
    return;
}

// 提取文本内容
List<String> texts = chunks.stream()
      .map(TextChunk::getContent)
      .toList();

// 从数据库获取分块内容
private List<TextChunk> fetchTextChunks(String fileMd5) {
  // 调用 Repository 查询数据
  List<DocumentVector> vectors = documentVectorRepository.findByFileMd5(fileMd5);

  // 转换为 TextChunk 列表
  return vectors.stream()
          .map(vector -> new TextChunk(
                  vector.getChunkId(),
                  vector.getTextContent()
          ))
          .toList();
}

// 调用外部模型生成向量
List<float[]> vectors = embeddingClient.embed(texts);

// 构建 Elasticsearch 文档并存储
List<EsDocument> esDocuments = IntStream.range(0, chunks.size())
  .mapToObj(i -> new EsDocument(
          UUID.randomUUID().toString(),
          fileMd5,
          chunks.get(i).getChunkId(),
          chunks.get(i).getContent(),
          vectors.get(i),
          "deepseek-embed", // 更新为 DeepSeek 的模型版本
          userId,
          orgTag,
          isPublic
  ))
  .toList();

elasticsearchService.bulkIndex(esDocuments); // 批量存储到 Elasticsearch

logger.info("向量化完成，fileMd5: {}", fileMd5);

import org.springframework.web.reactive.function.client.WebClient;

@Configuration
public class WebClientConfig {

    @Bean
    public WebClient webClient() {
        return WebClient.create("https://api.example.com");
    }
}

@Configuration
public class WebClientConfig {
  @Value("${embedding.api.url}")
  private String apiUrl;
  
  @Value("${embedding.api.key}")
  private String apiKey;
  
  @Bean
  public WebClient embeddingWebClient() {
    ExchangeStrategies strategies = ExchangeStrategies.builder()
        .codecs(configurer -> configurer
            .defaultCodecs()
            .maxInMemorySize(16 * 1024 * 1024)) // 16MB
        .build();

    return WebClient.builder()
        .baseUrl(apiUrl)
        .exchangeStrategies(strategies)
        .defaultHeader(HttpHeaders.AUTHORIZATION, "Bearer " + apiKey)
        .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
        .build();
  }
} 

// 嵌入向量生成客户端
@Component
public class EmbeddingClient {
  @Value("${embedding.api.model}")
  private String modelId;
  
  @Value("${embedding.api.batch-size:100}")
  private int batchSize;

  @Value("${embedding.api.dimension:2048}")
  private int dimension;
  
  private static final Logger logger = LoggerFactory.getLogger(EmbeddingClient.class);
  private final WebClient webClient;
  private final ObjectMapper objectMapper;

  public EmbeddingClient(WebClient embeddingWebClient, ObjectMapper objectMapper) {
      this.webClient = embeddingWebClient;
      this.objectMapper = objectMapper;
  }

  /**
   * 调用通义千问 API 生成向量
   * @param texts 输入文本列表
   * @return 对应的向量列表
   */
  public List<float[]> embed(List<String> texts) {
    try {
        logger.info("开始生成向量，文本数量: {}", texts.size());
        
        List<float[]> all = new ArrayList<>(texts.size());
        for (int start = 0; start < texts.size(); start += batchSize) {
            int end = Math.min(start + batchSize, texts.size());
            List<String> sub = texts.subList(start, end);
            logger.debug("调用向量 API, 批次: {}-{} (size={})", start, end - 1, sub.size());
            String response = callApiOnce(sub);
            all.addAll(parseVectors(response));
        }
        logger.info("成功生成向量，总数量: {}", all.size());
        return all;
    } catch (Exception e) {
        logger.error("调用向量化 API 失败: {}", e.getMessage(), e);
        throw new RuntimeException("向量生成失败", e);
    }
  }

  private String callApiOnce(List<String> batch) {
    Map<String, Object> requestBody = new HashMap<>();
    requestBody.put("model", modelId);
    requestBody.put("input", batch);
    requestBody.put("dimension", dimension);  // 直接在根级别设置dimension
    requestBody.put("encoding_format", "float");  // 添加编码格式

    return webClient.post()
            .uri("/embeddings")
            .bodyValue(requestBody)
            .retrieve()
            .bodyToMono(String.class)
            .retryWhen(Retry.fixedDelay(3, Duration.ofSeconds(1))
                    .filter(e -> e instanceof WebClientResponseException))
            .block(Duration.ofSeconds(30));
  }

  private List<float[]> parseVectors(String response) throws Exception {
      JsonNode jsonNode = objectMapper.readTree(response);
      JsonNode data = jsonNode.get("data");  // 兼容模式下使用data字段
      if (data == null || !data.isArray()) {
          throw new RuntimeException("API 响应格式错误: data 字段不存在或不是数组");
      }
      
      List<float[]> vectors = new ArrayList<>();
      for (JsonNode item : data) {
          JsonNode embedding = item.get("embedding");
          if (embedding != null && embedding.isArray()) {
              float[] vector = new float[embedding.size()];
              for (int i = 0; i < embedding.size(); i++) {
                  vector[i] = (float) embedding.get(i).asDouble();
              }
              vectors.add(vector);
          }
      }
      return vectors;
  }
}

// Java 对象
User user = new User("Gemini", 18);

// 翻译
String json = objectMapper.writeValueAsString(user);

// 结果
// {"name":"Gemini", "age":18}

String json = "{\"name\":\"Gemini\", \"age\":18}";

// 翻译
User user = objectMapper.readValue(json, User.class);

// 结果：user.getName() 等于 "Gemini"

List<EsDocument> esDocuments = IntStream.range(0, chunks.size())
  .mapToObj(i -> new EsDocument(
          UUID.randomUUID().toString(),
          fileMd5,
          chunks.get(i).getChunkId(),
          chunks.get(i).getContent(),
          vectors.get(i),
          "deepseek-embed", // 更新为 DeepSeek 的模型版本
          userId,
          orgTag,
          isPublic
  ))
  .toList();

// Elasticsearch客户端配置类
@Configuration
public class EsConfig {
  @Value("${elasticsearch.host}")
  private String host;

  @Value("${elasticsearch.port}")
  private int port;

  @Value("${elasticsearch.scheme:https}")
  private String scheme;

  @Value("${elasticsearch.username:elastic}")
  private String username;

  @Value("${elasticsearch.password:changeme}")
  private String password;

  @Bean
  public ElasticsearchClient elasticsearchClient() {
    // 创建低级客户端
    RestClientBuilder builder = RestClient.builder(new HttpHost(host, port, scheme));

    // 设置基本认证
    if (username != null && !username.isEmpty()) {
        BasicCredentialsProvider credsProvider = new BasicCredentialsProvider();
        credsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));
        builder.setHttpClientConfigCallback(httpClientBuilder -> {
            // 忽略 TLS 证书（仅限开发环境）
            try {
                SSLContext sslContext = SSLContexts.custom()
                        .loadTrustMaterial(null, (X509Certificate[] chain, String authType) -> true)
                        .build();
                httpClientBuilder.setSSLContext(sslContext);
                httpClientBuilder.setSSLHostnameVerifier(NoopHostnameVerifier.INSTANCE);
            } catch (Exception e) {
                // ignore
            }
            return httpClientBuilder.setDefaultCredentialsProvider(credsProvider);
        });
    }

    RestClient restClient = builder.build();

    // 创建传输层
    // 这里指定了 JacksonJsonpMapper，意味着 ES 客户端会复用 Spring Boot 默认的 Jackson 库来处理 JSON，保持项目依赖的一致性
    ElasticsearchTransport transport = new RestClientTransport(
            restClient, new JacksonJsonpMapper()
    );

    // 返回高级客户端
    return new ElasticsearchClient(transport);
  }
}

// 将文档列表转换为批量操作列表，每个文档都对应一个索引操作
List<BulkOperation> bulkOperations = documents.stream()
  .map(doc -> BulkOperation.of(op -> op.index(idx -> idx
          .index("knowledge_base") // 指定索引名称
          .id(doc.getId()) // 使用文档的ID作为Elasticsearch中的文档ID
          .document(doc) // 将文档对象作为数据源
  )))
  .toList();

BulkRequest request = BulkRequest.of(b -> b.operations(bulkOperations));

// 执行批量索引操作
BulkResponse response = esClient.bulk(request);

// 检查响应结果
if (response.errors()) {
    logger.error("批量索引过程中发生错误:");
    for (BulkResponseItem item : response.items()) {
        if (item.error() != null) {
            logger.error("文档索引失败 - ID: {}, 错误: {}", item.id(), item.error().reason());
        }
    }
    throw new RuntimeException("批量索引部分失败，请检查日志");
} else {
    logger.info("批量索引成功完成，文档数量: {}", documents.size());
}

@Component
public class EsIndexInitializer implements CommandLineRunner {

  private static final Logger logger = LoggerFactory.getLogger(EsIndexInitializer.class);

  @Autowired
  private ElasticsearchClient esClient;

  @Value("classpath:es-mappings/knowledge_base.json") // 加载 JSON 文件
  private org.springframework.core.io.Resource mappingResource;

  @Override
  public void run(String... args) throws Exception {
      try {
          initializeIndex();
      } catch (Exception exception) {
          // 特别处理连接关闭异常，尝试重新连接
          if (exception instanceof ConnectionClosedException || (exception.getCause() != null && exception.getCause() instanceof ConnectionClosedException)) {
              logger.error("Elasticsearch连接已关闭，等待5秒后重试...");
              try {
                  Thread.sleep(5000); // 等待5秒后重试
                  // 重新尝试初始化索引
                  initializeIndex();
              } catch (Exception retryException) {
                  logger.error("重试初始化索引失败，请检查Elasticsearch连接配置，比如说是否开启了 HTTPS 模式: {}", retryException.getMessage());
                  throw new RuntimeException("初始化索引失败，重试也未能成功", retryException);
              }
          } else {
              throw new RuntimeException("初始化索引失败", exception);
          }
      }
  }

  /**
   * 初始化索引的核心逻辑
   * @throws Exception
   */
  private void initializeIndex() throws Exception {
      // 检查索引是否存在
      BooleanResponse existsResponse = esClient.indices().exists(ExistsRequest.of(e -> e.index("knowledge_base")));
      if (!existsResponse.value()) {
          createIndex();
      } else {
          logger.info("索引 'knowledge_base' 已存在");
      }
  }

  /**
   * 创建索引
   * @throws Exception
   */
  private void createIndex() throws Exception {
      // 读取 JSON 文件内容
      String mappingJson = new String(Files.readAllBytes(mappingResource.getFile().toPath()), StandardCharsets.UTF_8);

      // 创建索引并应用映射
      CreateIndexRequest createIndexRequest = CreateIndexRequest.of(c -> c
              .index("knowledge_base") // 索引名称
              .withJson(new StringReader(mappingJson)) // 使用 JSON 文件定义映射
      );
      esClient.indices().create(createIndexRequest);
      logger.info("索引 'knowledge_base' 已创建");
  }
}