AI-чатбот на Java с Langchain4j и MongoDB Atlas

ChatLanguageModel model = OpenAiChatModel.builder()
    .apiKey(System.getenv("OPENAI_API_KEY"))
    .modelName("gpt-4")
    .temperature(0.7)
    .build();
 
String response = model.generate("Объясни, что такое векторные embedding'и");

EmbeddingModel embeddingModel = OpenAiEmbeddingModel.builder()
    .apiKey(apiKey)
    .modelName(OpenAiEmbeddingModelName.TEXT_EMBEDDING_3_SMALL)
    .build();
 
Response<Embedding> embeddingResponse = embeddingModel.embed("Java Spring Boot разработка");
float[] vector = embeddingResponse.content().vector();
// Получаем массив из 1536 float'ов для модели text-embedding-3-small

ChatMemory chatMemory = MessageWindowChatMemory.withMaxMessages(10);
 
chatMemory.add(UserMessage.from("Как установить Spring Boot?"));
chatMemory.add(AiMessage.from("Spring Boot устанавливается через Maven или Gradle..."));
chatMemory.add(UserMessage.from("А какую версию выбрать?"));
 
// Модель видит весь контекст беседы

PromptTemplate promptTemplate = PromptTemplate.from(
    """
    Ты - эксперт по {{topic}}.
    Ответь на вопрос пользователя, используя только факты.
    Если не уверен - так и скажи.
    
    Вопрос: {{question}}
    """
);
 
Map<String, Object> variables = new HashMap<>();
variables.put("topic", "Java concurrency");
variables.put("question", "Чем volatile отличается от synchronized?");
 
Prompt prompt = promptTemplate.apply(variables);
String response = model.generate(prompt.text());

@Data
class CodeReview {
    String summary;        // Краткое описание проблемы
    List<String> issues;   // Найденные баги
    int severity;          // Критичность от 1 до 10
}
 
AiService<CodeReviewService> service = AiServices.builder(CodeReviewService.class)
    .chatLanguageModel(model)
    .build();
 
CodeReview review = service.reviewCode(javaCodeString);
// Получаем строго типизированный объект вместо сырого текста

@Tool("Получает текущую температуру для указанного города")
public String getCurrentTemperature(@P("Название города") String city) {
    // Реальный вызов API погоды
    return weatherService.getTemperature(city);
}
 
@Tool("Вычисляет квадратный корень числа")  
public double sqrt(@P("Число для вычисления") double number) {
    return Math.sqrt(number);
}

ConversationalRetrievalChain chain = ConversationalRetrievalChain.builder()
    .chatLanguageModel(model)
    .retriever(documentRetriever)  // Компонент поиска по базе
    .chatMemory(chatMemory)
    .build();
 
String answer = chain.execute("Как настроить MongoDB Atlas?");

model.generate("Напиши подробную статью про JVM", new StreamingResponseHandler<String>() {
    @Override
    public void onNext(String token) {
        System.out.print(token);  // Выводим по мере поступления
    }
    
    @Override
    public void onComplete(Response<String> response) {
        System.out.println("\n[Готово]");
    }
    
    @Override
    public void onError(Throwable error) {
        logger.error("Ошибка генерации", error);
    }
});

ChatLanguageModel resilientModel = OpenAiChatModel.builder()
    .apiKey(apiKey)
    .timeout(Duration.ofSeconds(30))
    .maxRetries(3)  // Автоматические повторы
    .logRequests(true)
    .logResponses(true)
    .build();

MongoDatabase database = mongoClient.getDatabase("chatbot_db");
MongoCollection<Document> collection = database.getCollection("knowledge_base");
 
// Структура документа
Document doc = new Document()
    .append("_id", new ObjectId())
    .append("content", "Текст статьи про MongoDB")  
    .append("embedding", embeddingVector)  // float[] преобразованный в List<Double>
    .append("metadata", new Document()
        .append("source", "confluence")
        .append("author", "ivanov")
        .append("created_at", new Date())
    );
 
collection.insertOne(doc);

List<Document> pipeline = Arrays.asList(
    new Document("$vectorSearch", 
        new Document("index", "vector_index")
            .append("path", "embedding")  
            .append("queryVector", queryEmbeddingList)
            .append("numCandidates", 150)  // Сколько документов проверять
            .append("limit", 10)           // Сколько вернуть
    )
);
 
AggregateIterable<Document> results = collection.aggregate(pipeline);

List<Document> batch = new ArrayList<>();
for (Article article : articles) {
    Embedding embedding = embeddingModel.embed(article.getContent()).content();
    
    Document doc = new Document()
        .append("content", article.getContent())
        .append("embedding", embedding.vectorAsList())
        .append("metadata", article.getMetadata());
    
    batch.add(doc);
    
    if (batch.size() >= 100) {
        collection.insertMany(batch);
        batch.clear();
    }
}
 
if (!batch.isEmpty()) {
    collection.insertMany(batch);  // Не забываем остаток
}

// Разбиваем текст на фрагменты с перекрытием
int chunkSize = 1000;  // Токенов
int overlap = 100;
 
List<String> chunks = splitIntoChunks(article.getContent(), chunkSize, overlap);
 
for (int i = 0; i < chunks.size(); i++) {
    Embedding embedding = embeddingModel.embed(chunks.get(i)).content();
    
    Document chunkDoc = new Document()
        .append("content", chunks.get(i))
        .append("embedding", embedding.vectorAsList())  
        .append("metadata", new Document()
            .append("article_id", article.getId())
            .append("chunk_index", i)
            .append("total_chunks", chunks.size())
        );
    
    collection.insertOne(chunkDoc);
}

<dependency>
    <groupId>dev.langchain4j</groupId>
    <artifactId>langchain4j</artifactId>
    <version>0.27.1</version>
</dependency>
 
<dependency>
    <groupId>dev.langchain4j</groupId>
    <artifactId>langchain4j-open-ai</artifactId>
    <version>0.27.1</version>
</dependency>
 
<dependency>
    <groupId>dev.langchain4j</groupId>
    <artifactId>langchain4j-embeddings-all-minilm-l6-v2</artifactId>
    <version>0.27.1</version>
</dependency>
 
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
    <version>3.2.3</version>
</dependency>
 
<dependency>
    <groupId>org.mongodb</groupId>
    <artifactId>mongodb-driver-sync</artifactId>
    <version>4.11.1</version>
</dependency>

@Configuration
public class ChatbotConfig {
    
    @Value("${openai.api-key}")
    private String apiKey;
    
    @Value("${openai.model}")
    private String modelName;
    
    @Value("${openai.temperature}")
    private Double temperature;
    
    @Value("${mongodb.uri}")
    private String mongoUri;
    
    @Bean
    public MongoClient mongoClient() {
        return MongoClients.create(mongoUri);
    }
    
    @Bean
    public ChatLanguageModel chatModel() {
        return OpenAiChatModel.builder()
            .apiKey(apiKey)
            .modelName(modelName)
            .temperature(temperature)
            .timeout(Duration.ofSeconds(60))
            .maxRetries(3)
            .logRequests(false)  // В проде лучше выключить - треш в логах
            .logResponses(false)
            .build();
    }
    
    @Bean
    public EmbeddingModel embeddingModel() {
        return OpenAiEmbeddingModel.builder()
            .apiKey(apiKey)
            .modelName(OpenAiEmbeddingModelName.TEXT_EMBEDDING_3_SMALL)
            .build();
    }
    
    @Bean
    public EmbeddingStore<TextSegment> embeddingStore(
            MongoClient mongoClient,
            @Value("${mongodb.database}") String dbName,
            @Value("${mongodb.collection}") String collectionName) {
        
        return MongoDbEmbeddingStore.builder()
            .mongoClient(mongoClient)
            .databaseName(dbName)
            .collectionName(collectionName)
            .indexName("vector_index")
            .createIndex(false)  // Индекс создаём вручную через UI
            .build();
    }
    
    @Bean
    public ChatMemory chatMemory(@Value("${chat.memory-window}") int windowSize) {
        return MessageWindowChatMemory.withMaxMessages(windowSize);
    }
}

@Service
public class DocumentRetrievalService {
    
    private final ContentRetriever contentRetriever;
    
    public DocumentRetrievalService(
            EmbeddingStore<TextSegment> embeddingStore,
            EmbeddingModel embeddingModel) {
        
        this.contentRetriever = EmbeddingStoreContentRetriever.builder()
            .embeddingStore(embeddingStore)
            .embeddingModel(embeddingModel)
            .maxResults(5)
            .minScore(0.75)  // Порог релевантности
            .build();
    }
    
    public List<Content> retrieve(String query) {
        Query queryObj = Query.from(query);
        return contentRetriever.retrieve(queryObj);
    }
}

@Service
public class ChatbotService {
    
    private final ChatLanguageModel chatModel;
    private final DocumentRetrievalService retrievalService;
    private final ChatMemory chatMemory;
    
    @Value("${chat.system-message}")
    private String systemMessage;
    
    public ChatbotService(
            ChatLanguageModel chatModel,
            DocumentRetrievalService retrievalService,
            ChatMemory chatMemory) {
        this.chatModel = chatModel;
        this.retrievalService = retrievalService;
        this.chatMemory = chatMemory;
    }
    
    public String chat(String userMessage) {
        // Достаём релевантные документы
        List<Content> relevantDocs = retrievalService.retrieve(userMessage);
        
        // Формируем контекст из найденных документов
        String context = buildContext(relevantDocs);
        
        // Создаём промпт с контекстом
        String augmentedPrompt = buildPrompt(context, userMessage);
        
        // Добавляем в историю
        chatMemory.add(UserMessage.from(augmentedPrompt));
        
        // Генерируем ответ
        Response<AiMessage> response = chatModel.generate(
            chatMemory.messages()
        );
        
        // Сохраняем ответ в историю
        chatMemory.add(response.content());
        
        return response.content().text();
    }
    
    private String buildContext(List<Content> docs) {
        if (docs.isEmpty()) {
            return "Релевантных документов не найдено.";
        }
        
        StringBuilder context = new StringBuilder("Контекст из базы знаний:\n\n");
        for (int i = 0; i < docs.size(); i++) {
            context.append(String.format("[Документ %d]\n", i + 1));
            context.append(docs.get(i).textSegment().text());
            context.append("\n\n");
        }
        return context.toString();
    }
    
    private String buildPrompt(String context, String question) {
        return String.format("""
            %s
            
            %s
            
            Вопрос пользователя: %s
            
            Ответь на вопрос, используя информацию из контекста.
            Если контекст не содержит нужной информации - скажи об этом.
            """, systemMessage, context, question);
    }
}

@RestController
@RequestMapping("/api/chat")
public class ChatController {
    
    private final ChatbotService chatbotService;
    
    public ChatController(ChatbotService chatbotService) {
        this.chatbotService = chatbotService;
    }
    
    @PostMapping
    public ResponseEntity<ChatResponse> chat(@RequestBody ChatRequest request) {
        String answer = chatbotService.chat(request.getMessage());
        return ResponseEntity.ok(new ChatResponse(answer));
    }
    
    @PostMapping("/stream")
    public SseEmitter chatStream(@RequestBody ChatRequest request) {
        SseEmitter emitter = new SseEmitter(60_000L);
        
        // Асинхронная обработка в отдельном потоке
        CompletableFuture.runAsync(() -> {
            try {
                streamResponse(request.getMessage(), emitter);
                emitter.complete();
            } catch (Exception e) {
                emitter.completeWithError(e);
            }
        });
        
        return emitter;
    }
    
    private void streamResponse(String message, SseEmitter emitter) {
        // Реализация streaming'а - об этом дальше
    }
}

@Data
@NoArgsConstructor
@AllArgsConstructor
public class ChatRequest {
    private String message;
    private String sessionId;  // Для multi-user поддержки
}
 
@Data
@AllArgsConstructor
public class ChatResponse {
    private String response;
    private long timestamp = System.currentTimeMillis();
}

@Configuration
public class CacheConfig {
    
    @Bean
    public Cache<String, Embedding> embeddingCache() {
        return Caffeine.newBuilder()
            .maximumSize(1000)
            .expireAfterWrite(Duration.ofHours(1))
            .build();
    }
}

private final Cache<String, Embedding> embeddingCache;
 
public List<Content> retrieve(String query) {
    Embedding queryEmbedding = embeddingCache.get(query, 
        key -> embeddingModel.embed(key).content()
    );
    
    List<EmbeddingMatch<TextSegment>> matches = embeddingStore.findRelevant(
        queryEmbedding, 5, 0.75
    );
    
    return matches.stream()
        .map(match -> Content.from(match.embedded()))
        .collect(Collectors.toList());
}

@Service
public class SessionAwareChatMemoryProvider {
    
    private final Map<String, ChatMemory> sessionMemories = new ConcurrentHashMap<>();
    private final int maxMessages;
    
    public SessionAwareChatMemoryProvider(@Value("${chat.memory-window}") int maxMessages) {
        this.maxMessages = maxMessages;
    }
    
    public ChatMemory getMemory(String sessionId) {
        return sessionMemories.computeIfAbsent(
            sessionId, 
            id -> MessageWindowChatMemory.withMaxMessages(maxMessages)
        );
    }
    
    public void clearMemory(String sessionId) {
        sessionMemories.remove(sessionId);
    }
    
    @Scheduled(fixedRate = 3600000)  // Каждый час
    public void cleanupInactiveSessions() {
        // Удаляем старые сессии без активности
        // В реальности нужно хранить timestamp последнего обращения
    }
}

public String chat(String userMessage, String sessionId) {
    ChatMemory memory = memoryProvider.getMemory(sessionId);
    
    List<Content> relevantDocs = retrievalService.retrieve(userMessage);
    String context = buildContext(relevantDocs);
    String augmentedPrompt = buildPrompt(context, userMessage);
    
    memory.add(UserMessage.from(augmentedPrompt));
    
    Response<AiMessage> response = chatModel.generate(memory.messages());
    memory.add(response.content());
    
    return response.content().text();
}

@Service
public class TokenAwareChatService {
  
  private final Encoding encoding;
  private final int maxContextTokens = 6000;  // Оставляем запас
  
  public TokenAwareChatService() {
      EncodingRegistry registry = Encodings.newDefaultEncodingRegistry();
      this.encoding = registry.getEncoding(EncodingType.CL100K_BASE);
  }
  
  public String chat(String userMessage, String sessionId) {
      ChatMemory memory = memoryProvider.getMemory(sessionId);
      
      // Формируем промпт
      List<Content> docs = retrievalService.retrieve(userMessage);
      String context = buildContext(docs);
      String prompt = buildPrompt(context, userMessage);
      
      // Проверяем текущий размер контекста
      int currentTokens = countTokens(memory.messages());
      int newMessageTokens = countTokens(prompt);
      
      // Если превышаем лимит - обрезаем историю
      if (currentTokens + newMessageTokens > maxContextTokens) {
          trimMemory(memory, maxContextTokens - newMessageTokens);
      }
      
      memory.add(UserMessage.from(prompt));
      
      Response<AiMessage> response = chatModel.generate(memory.messages());
      memory.add(response.content());
      
      return response.content().text();
  }
  
  private int countTokens(List<ChatMessage> messages) {
      int total = 0;
      for (ChatMessage msg : messages) {
          total += encoding.countTokens(msg.text());
      }
      return total;
  }
  
  private void trimMemory(ChatMemory memory, int targetTokens) {
      List<ChatMessage> messages = new ArrayList<>(memory.messages());
      
      // Всегда сохраняем системное сообщение
      ChatMessage systemMsg = messages.get(0);
      messages.remove(0);
      
      // Удаляем старые сообщения, пока не влезем в лимит
      while (countTokens(messages) > targetTokens && messages.size() > 2) {
          messages.remove(0);  // Удаляем самое старое
          messages.remove(0);  // И его ответ
      }
      
      // Пересоздаём память
      memory.clear();
      memory.add(systemMsg);
      messages.forEach(memory::add);
  }
}

private String summarizeOldMessages(List<ChatMessage> messages) {
    if (messages.size() < 6) {
        return null;  // Недостаточно истории для суммаризации
    }
    
    // Берём первые N сообщений (кроме системного)
    List<ChatMessage> toSummarize = messages.subList(1, Math.min(6, messages.size()));
    
    String dialogText = toSummarize.stream()
        .map(msg -> {
            String role = msg instanceof UserMessage ? "User" : "Assistant";
            return role + ": " + msg.text();
        })
        .collect(Collectors.joining("\n"));
    
    String summaryPrompt = String.format("""
        Создай краткое резюме следующего диалога, сохранив ключевые факты и контекст:
        
        %s
        
        Резюме должно быть компактным, но информативным.
        """, dialogText);
    
    Response<AiMessage> summary = chatModel.generate(summaryPrompt);
    return summary.content().text();
}

@PostMapping("/stream")
public SseEmitter chatStream(@RequestBody ChatRequest request) {
    SseEmitter emitter = new SseEmitter(120_000L);
    String sessionId = request.getSessionId();
    
    executorService.execute(() -> {
        try {
            ChatMemory memory = memoryProvider.getMemory(sessionId);
            
            // Получаем контекст
            List<Content> docs = retrievalService.retrieve(request.getMessage());
            String context = buildContext(docs);
            String prompt = buildPrompt(context, request.getMessage());
            
            memory.add(UserMessage.from(prompt));
            
            // Стримим ответ токен за токеном
            StringBuilder fullResponse = new StringBuilder();
            
            chatModel.generate(memory.messages(), new StreamingResponseHandler<AiMessage>() {
                @Override
                public void onNext(String token) {
                    try {
                        fullResponse.append(token);
                        emitter.send(SseEmitter.event()
                            .data(token)
                            .name("token"));
                    } catch (IOException e) {
                        emitter.completeWithError(e);
                    }
                }
                
                @Override
                public void onComplete(Response<AiMessage> response) {
                    memory.add(response.content());
                    emitter.complete();
                }
                
                @Override
                public void onError(Throwable error) {
                    logger.error("Streaming error", error);
                    emitter.completeWithError(error);
                }
            });
            
        } catch (Exception e) {
            emitter.completeWithError(e);
        }
    });
    
    return emitter;
}

emitter.onTimeout(() -> {
    logger.warn("SSE timeout for session: {}", sessionId);
    cleanupIncompleteResponse(sessionId);
});
 
emitter.onCompletion(() -> {
    logger.debug("SSE completed for session: {}", sessionId);
});
 
private void cleanupIncompleteResponse(String sessionId) {
    ChatMemory memory = memoryProvider.getMemory(sessionId);
    List<ChatMessage> messages = new ArrayList<>(memory.messages());
    
    // Если последнее сообщение - от пользователя, и нет ответа - удаляем
    if (!messages.isEmpty() && messages.get(messages.size() - 1) instanceof UserMessage) {
        messages.remove(messages.size() - 1);
        memory.clear();
        messages.forEach(memory::add);
    }
}

@Configuration
public class RateLimiterConfig {
    
    @Bean
    public RateLimiter openAiRateLimiter() {
        RateLimiterConfig config = RateLimiterConfig.custom()
            .limitForPeriod(50)
            .limitRefreshPeriod(Duration.ofMinutes(1))
            .timeoutDuration(Duration.ofSeconds(5))
            .build();
        
        return RateLimiter.of("openai", config);
    }
}
 
// В сервисе
@Autowired
private RateLimiter rateLimiter;
 
public String chat(String message, String sessionId) {
    return Try.ofSupplier(
        RateLimiter.decorateSupplier(rateLimiter, () -> 
            chatInternal(message, sessionId)
        )
    ).getOrElseThrow(ex -> 
        new RateLimitExceededException("Too many requests", ex)
    );
}

@Bean
public CircuitBreaker chatCircuitBreaker() {
    CircuitBreakerConfig config = CircuitBreakerConfig.custom()
        .failureRateThreshold(50)
        .waitDurationInOpenState(Duration.ofSeconds(30))
        .slidingWindowSize(10)
        .build();
    
    return CircuitBreaker.of("chat", config);
}

public List<Content> retrieveWithFilters(
        String query, 
        Map<String, Object> filters) {
    
    Embedding queryEmbedding = embeddingModel.embed(query).content();
    List<Double> vectorList = Arrays.stream(queryEmbedding.vector())
        .boxed()
        .collect(Collectors.toList());
    
    List<Bson> pipeline = new ArrayList<>();
    
    // Векторный поиск
    Document vectorSearch = new Document("$vectorSearch",
        new Document("index", "vector_index")
            .append("path", "embedding")
            .append("queryVector", vectorList)
            .append("numCandidates", 100)
            .append("limit", 20));  // Берём больше кандидатов
    
    pipeline.add(vectorSearch);
    
    // Добавляем фильтры по метаданным
    if (!filters.isEmpty()) {
        Document matchStage = new Document();
        
        if (filters.containsKey("author")) {
            matchStage.append("metadata.author", filters.get("author"));
        }
        
        if (filters.containsKey("dateFrom")) {
            matchStage.append("metadata.created_at", 
                new Document("$gte", filters.get("dateFrom")));
        }
        
        if (filters.containsKey("tags")) {
            @SuppressWarnings("unchecked")
            List<String> tags = (List<String>) filters.get("tags");
            matchStage.append("metadata.tags", 
                new Document("$in", tags));
        }
        
        pipeline.add(new Document("$match", matchStage));
    }
    
    // Ограничиваем финальный результат
    pipeline.add(new Document("$limit", 5));
    
    MongoCollection<Document> collection = mongoClient
        .getDatabase(databaseName)
        .getCollection(collectionName);
    
    List<Content> results = new ArrayList<>();
    collection.aggregate(pipeline).forEach(doc -> {
        String text = doc.getString("content");
        Document metadata = doc.get("metadata", Document.class);
        results.add(Content.from(text, metadata));
    });
    
    return results;
}

public List<Content> retrieveWithAdaptiveFilters(
        String query,
        Map<String, Object> strictFilters,
        Map<String, Object> softFilters) {
    
    // Пробуем со строгими фильтрами
    List<Content> results = retrieveWithFilters(query, strictFilters);
    
    if (results.size() >= 3) {
        return results;
    }
    
    // Если мало результатов - смягчаем фильтры
    logger.debug("Недостаточно результатов с строгими фильтрами, применяем мягкие");
    
    Map<String, Object> relaxedFilters = new HashMap<>(softFilters);
    results = retrieveWithFilters(query, relaxedFilters);
    
    if (results.size() >= 3) {
        return results;
    }
    
    // В крайнем случае - только векторный поиск
    logger.debug("Поиск без фильтров метаданных");
    return retrieveWithFilters(query, Collections.emptyMap());
}

@Service
public class RerankingService {
    
    private final ChatLanguageModel model;
    
    public List<Content> rerank(String query, List<Content> candidates) {
        if (candidates.size() <= 3) {
            return candidates;  // Не тратим время на малое количество
        }
        
        List<ScoredContent> scored = new ArrayList<>();
        
        for (Content content : candidates) {
            double score = calculateRelevanceScore(query, content.text());
            scored.add(new ScoredContent(content, score));
        }
        
        // Сортируем по убыванию релевантности
        scored.sort(Comparator.comparingDouble(
            ScoredContent::score).reversed());
        
        return scored.stream()
            .limit(5)
            .map(ScoredContent::content)
            .collect(Collectors.toList());
    }
    
    private double calculateRelevanceScore(String query, String document) {
        String prompt = String.format("""
            Оцени релевантность документа для данного запроса.
            Верни только число от 0 до 1, где 0 - совершенно нерелевантен, 1 - идеально подходит.
            
            Запрос: %s
            
            Документ: %s
            
            Оценка:
            """, query, document.substring(0, Math.min(500, document.length())));
        
        String response = model.generate(prompt);
        
        try {
            return Double.parseDouble(response.trim());
        } catch (NumberFormatException e) {
            logger.warn("Не удалось распарсить оценку: {}", response);
            return 0.5;  // Средняя оценка по умолчанию
        }
    }
    
    private record ScoredContent(Content content, double score) {}
}

public List<Content> hybridSearch(String query) {
    // Векторный поиск
    List<ScoredContent> vectorResults = vectorSearch(query);
    
    // Полнотекстовый поиск MongoDB
    List<ScoredContent> textResults = textSearch(query);
    
    // Объединяем с весами
    Map<String, Double> combinedScores = new HashMap<>();
    
    double vectorWeight = 0.7;
    double textWeight = 0.3;
    
    for (ScoredContent result : vectorResults) {
        String id = result.content().id();
        combinedScores.put(id, 
            combinedScores.getOrDefault(id, 0.0) + result.score() * vectorWeight);
    }
    
    for (ScoredContent result : textResults) {
        String id = result.content().id();
        combinedScores.put(id,
            combinedScores.getOrDefault(id, 0.0) + result.score() * textWeight);
    }
    
    // Сортируем по комбинированному скору
    return combinedScores.entrySet().stream()
        .sorted(Map.Entry.<String, Double>comparingByValue().reversed())
        .limit(5)
        .map(entry -> getContentById(entry.getKey()))
        .collect(Collectors.toList());
}
 
private List<ScoredContent> textSearch(String query) {
    MongoCollection<Document> collection = mongoClient
        .getDatabase(databaseName)
        .getCollection(collectionName);
    
    Bson textSearchStage = Aggregates.match(
        Filters.text(query)
    );
    
    List<ScoredContent> results = new ArrayList<>();
    collection.aggregate(Arrays.asList(
        textSearchStage,
        Aggregates.limit(10)
    )).forEach(doc -> {
        String text = doc.getString("content");
        Double score = doc.getDouble("score");
        results.add(new ScoredContent(Content.from(text), score));
    });
    
    return results;
}

private List<Content> expandContext(Content content, int windowSize) {
    Document metadata = content.metadata();
    String articleId = metadata.getString("article_id");
    int chunkIndex = metadata.getInteger("chunk_index");
    
    // Достаём соседние чанки
    MongoCollection<Document> collection = mongoClient
        .getDatabase(databaseName)
        .getCollection(collectionName);
    
    List<Content> expanded = new ArrayList<>();
    
    for (int i = chunkIndex - windowSize; i <= chunkIndex + windowSize; i++) {
        if (i < 0) continue;
        
        Document chunk = collection.find(
            Filters.and(
                Filters.eq("metadata.article_id", articleId),
                Filters.eq("metadata.chunk_index", i)
            )
        ).first();
        
        if (chunk != null) {
            expanded.add(Content.from(chunk.getString("content")));
        }
    }
    
    return expanded;
}

public List<String> chunkByLength(String text, int chunkSize) {
    List<String> chunks = new ArrayList<>();
    for (int i = 0; i < text.length(); i += chunkSize) {
        chunks.add(text.substring(i, Math.min(i + chunkSize, text.length())));
    }
    return chunks;
}

public List<String> semanticChunk(String text) {
    // Разбиваем на абзацы
    String[] paragraphs = text.split("
 
+");
    
    List<String> chunks = new ArrayList<>();
    StringBuilder currentChunk = new StringBuilder();
    int currentTokens = 0;
    
    for (String para : paragraphs) {
        int paraTokens = tokenizer.countTokens(para);
        
        if (currentTokens + paraTokens > 800) {
            // Текущий чанк заполнен
            if (currentChunk.length() > 0) {
                chunks.add(currentChunk.toString());
                currentChunk = new StringBuilder();
                currentTokens = 0;
            }
        }
        
        currentChunk.append(para).append("
 
");
        currentTokens += paraTokens;
    }
    
    if (currentChunk.length() > 0) {
        chunks.add(currentChunk.toString());
    }
    
    return chunks;
}

public List<String> chunkWithOverlap(String text, int size, int overlap) {
    List<String> sentences = splitIntoSentences(text);
    List<String> chunks = new ArrayList<>();
    
    int start = 0;
    while (start < sentences.size()) {
        int end = Math.min(start + size, sentences.size());
        
        String chunk = String.join(" ", sentences.subList(start, end));
        chunks.add(chunk);
        
        start += (size - overlap);
    }
    
    return chunks;
}

@Service  
public class CachedEmbeddingService {
    
    private final LoadingCache<String, Embedding> memoryCache;
    private final RedisTemplate<String, byte[]> redisCache;
    private final EmbeddingModel model;
    
    public CachedEmbeddingService(EmbeddingModel model) {
        this.model = model;
        
        this.memoryCache = Caffeine.newBuilder()
            .maximumSize(1000)
            .expireAfterWrite(Duration.ofHours(1))
            .build(this::generateEmbedding);
    }
    
    public Embedding getEmbedding(String text) {
        // Уровень 1: память процесса
        try {
            return memoryCache.get(text);
        } catch (Exception e) {
            logger.error("Cache error", e);
            return generateEmbedding(text);
        }
    }
    
    private Embedding generateEmbedding(String text) {
        // Уровень 2: Redis
        String cacheKey = "emb:" + hashText(text);
        byte[] cached = redisCache.opsForValue().get(cacheKey);
        
        if (cached != null) {
            return deserializeEmbedding(cached);
        }
        
        // Уровень 3: реальная генерация
        Embedding embedding = model.embed(text).content();
        
        // Сохраняем в Redis
        redisCache.opsForValue().set(
            cacheKey, 
            serializeEmbedding(embedding),
            Duration.ofDays(7)
        );
        
        return embedding;
    }
    
    private String hashText(String text) {
        return DigestUtils.sha256Hex(text);
    }
}

public Map<String, Embedding> batchEmbed(List<String> texts) {
    Map<String, Embedding> results = new HashMap<>();
    
    // Разбиваем на батчи по 100 (безопасный лимит)
    List<List<String>> batches = Lists.partition(texts, 100);
    
    for (List<String> batch : batches) {
        Response<List<Embedding>> response = model.embedAll(batch);
        
        for (int i = 0; i < batch.size(); i++) {
            results.put(batch.get(i), response.content().get(i));
        }
    }
    
    return results;
}

public List<Content> optimizedVectorSearch(String query, int topK) {
    Embedding queryEmbedding = embeddingCache.get(query,
        key -> embeddingModel.embed(key).content()
    );
    
    // Эмпирическое правило: numCandidates = topK * 15
    int numCandidates = Math.min(topK * 15, 200);
    
    Document vectorSearchStage = new Document("$vectorSearch",
        new Document("index", "vector_index")
            .append("path", "embedding")
            .append("queryVector", queryEmbedding.vectorAsList())
            .append("numCandidates", numCandidates)
            .append("limit", topK)
    );
    
    // Добавляем проекцию - не тянем лишние поля
    Document projectStage = new Document("$project",
        new Document("content", 1)
            .append("metadata", 1)
            .append("score", new Document("$meta", "vectorSearchScore"))
    );
    
    List<Bson> pipeline = Arrays.asList(
        vectorSearchStage,
        projectStage
    );
    
    long startTime = System.nanoTime();
    List<Content> results = executeSearch(pipeline);
    long duration = (System.nanoTime() - startTime) / 1_000_000;
    
    metrics.recordSearchLatency(duration);
    
    return results;
}

@Component
public class VectorIndexWarmer {
    
    private final DocumentRetrievalService retrievalService;
    
    @EventListener(ApplicationReadyEvent.class)
    public void warmupIndex() {
        logger.info("Прогреваем векторный индекс...");
        
        List<String> warmupQueries = Arrays.asList(
            "Java Spring Boot",
            "MongoDB настройка",
            "микросервисы архитектура",
            "REST API разработка",
            "Docker контейнеризация"
        );
        
        warmupQueries.parallelStream().forEach(query -> {
            try {
                retrievalService.retrieve(query);
            } catch (Exception e) {
                logger.warn("Warmup failed for: {}", query, e);
            }
        });
        
        logger.info("Индекс прогрет");
    }
}

public String chatWithParallelRetrieval(String message, String sessionId) {
    ExecutorService executor = Executors.newFixedThreadPool(2);
    
    try {
        // Параллельно: поиск документов и генерация query embedding
        Future<List<Content>> documentsFuture = executor.submit(() -> 
            retrievalService.retrieve(message)
        );
        
        Future<String> contextFuture = executor.submit(() ->
            buildAdditionalContext(sessionId)
        );
        
        List<Content> docs = documentsFuture.get(5, TimeUnit.SECONDS);
        String additionalContext = contextFuture.get(2, TimeUnit.SECONDS);
        
        String prompt = buildPrompt(docs, additionalContext, message);
        
        ChatMemory memory = memoryProvider.getMemory(sessionId);
        memory.add(UserMessage.from(prompt));
        
        Response<AiMessage> response = chatModel.generate(memory.messages());
        memory.add(response.content());
        
        return response.content().text();
        
    } catch (TimeoutException e) {
        logger.error("Retrieval timeout", e);
        throw new ServiceException("Search took too long");
    } catch (Exception e) {
        logger.error("Parallel retrieval failed", e);
        throw new ServiceException("Internal error");
    } finally {
        executor.shutdown();
    }
}

@Component
public class SearchMetrics {
    
    private final MeterRegistry registry;
    private final Timer searchTimer;
    private final Counter cacheHitCounter;
    private final Counter cacheMissCounter;
    
    public SearchMetrics(MeterRegistry registry) {
        this.registry = registry;
        this.searchTimer = Timer.builder("vector.search.duration")
            .description("Время выполнения векторного поиска")
            .register(registry);
        
        this.cacheHitCounter = Counter.builder("embedding.cache.hits")
            .description("Попадания в кеш эмбеддингов")
            .register(registry);
        
        this.cacheMissCounter = Counter.builder("embedding.cache.misses")
            .description("Промахи кеша эмбеддингов")
            .register(registry);
    }
    
    public void recordSearchLatency(long milliseconds) {
        searchTimer.record(milliseconds, TimeUnit.MILLISECONDS);
    }
    
    public void recordCacheHit() {
        cacheHitCounter.increment();
    }
    
    public void recordCacheMiss() {
        cacheMissCounter.increment();
    }
}

Document projectStage = new Document("$project",
    new Document("content", 1)
        .append("metadata.author", 1)
        .append("metadata.title", 1)
        .append("_id", 0)
        .append("embedding", 0)  // Явно исключаем вектор
);

chatbot-mongodb/
├── src/main/java/com/example/chatbot/
│   ├── config/
│   │   ├── MongoConfig.java
│   │   ├── OpenAiConfig.java
│   │   └── CacheConfig.java
│   ├── controller/
│   │   └── ChatController.java
│   ├── service/
│   │   ├── ChatService.java
│   │   ├── DocumentIngestionService.java
│   │   ├── EmbeddingService.java
│   │   └── RetrievalService.java
│   ├── repository/
│   │   └── VectorRepository.java
│   ├── model/
│   │   ├── ChatMessage.java
│   │   ├── Document.java
│   │   └── SearchResult.java
│   └── ChatbotApplication.java
├── src/main/resources/
│   ├── application.yml
│   └── prompts/
│       └── system-message.txt
└── pom.xml

@Configuration
public class MongoConfig {
    
    @Value("${mongodb.uri}")
    private String uri;
    
    @Value("${mongodb.database}")
    private String database;
    
    @Bean
    public MongoClient mongoClient() {
        MongoClientSettings settings = MongoClientSettings.builder()
            .applyConnectionString(new ConnectionString(uri))
            .applyToConnectionPoolSettings(builder -> 
                builder.maxSize(20)
                    .minSize(5)
                    .maxWaitTime(2, TimeUnit.SECONDS)
                    .maxConnectionIdleTime(30, TimeUnit.SECONDS)
            )
            .applyToSocketSettings(builder ->
                builder.connectTimeout(5, TimeUnit.SECONDS)
                    .readTimeout(10, TimeUnit.SECONDS)
            )
            .retryWrites(true)
            .build();
        
        return MongoClients.create(settings);
    }
    
    @Bean
    public MongoDatabase database(MongoClient client) {
        return client.getDatabase(database);
    }
}

@Service
public class DocumentIngestionService {
    
    private final EmbeddingService embeddingService;
    private final VectorRepository repository;
    private final Tokenizer tokenizer;
    
    private static final int CHUNK_SIZE = 800;
    private static final int OVERLAP = 100;
    
    public void ingestDocument(String title, String content, Map<String, Object> metadata) {
        List<String> chunks = chunkDocument(content);
        
        for (int i = 0; i < chunks.size(); i++) {
            String chunk = chunks.get(i);
            
            Embedding embedding = embeddingService.generateEmbedding(chunk);
            
            Map<String, Object> chunkMetadata = new HashMap<>(metadata);
            chunkMetadata.put("title", title);
            chunkMetadata.put("chunk_index", i);
            chunkMetadata.put("total_chunks", chunks.size());
            chunkMetadata.put("indexed_at", LocalDateTime.now());
            
            repository.save(chunk, embedding, chunkMetadata);
        }
        
        logger.info("Проиндексирован документ '{}': {} чанков", title, chunks.size());
    }
    
    private List<String> chunkDocument(String text) {
        String[] paragraphs = text.split("
 
+");
        List<String> chunks = new ArrayList<>();
        
        StringBuilder currentChunk = new StringBuilder();
        int currentTokens = 0;
        
        for (String paragraph : paragraphs) {
            int paraTokens = tokenizer.count(paragraph);
            
            if (currentTokens + paraTokens > CHUNK_SIZE && currentChunk.length() > 0) {
                chunks.add(currentChunk.toString().trim());
                
                // Добавляем overlap из последних предложений
                String overlap = extractOverlap(currentChunk.toString(), OVERLAP);
                currentChunk = new StringBuilder(overlap);
                currentTokens = tokenizer.count(overlap);
            }
            
            currentChunk.append(paragraph).append("
 
");
            currentTokens += paraTokens;
        }
        
        if (currentChunk.length() > 0) {
            chunks.add(currentChunk.toString().trim());
        }
        
        return chunks;
    }
    
    private String extractOverlap(String text, int tokens) {
        String[] sentences = text.split("\\. ");
        StringBuilder overlap = new StringBuilder();
        int count = 0;
        
        for (int i = sentences.length - 1; i >= 0 && count < tokens; i--) {
            String sentence = sentences[i];
            int sentTokens = tokenizer.count(sentence);
            
            if (count + sentTokens <= tokens) {
                overlap.insert(0, sentence + ". ");
                count += sentTokens;
            } else {
                break;
            }
        }
        
        return overlap.toString();
    }
}

@Service
public class ChatService {
    
    private final RetrievalService retrievalService;
    private final ChatLanguageModel chatModel;
    private final Map<String, ChatMemory> sessionMemories = new ConcurrentHashMap<>();
    
    @Value("${chat.max-context-tokens}")
    private int maxContextTokens;
    
    public ChatResponse chat(ChatRequest request) {
        String sessionId = request.getSessionId();
        ChatMemory memory = getOrCreateMemory(sessionId);
        
        // Поиск релевантных документов
        List<SearchResult> documents = retrievalService.search(
            request.getMessage(),
            5,
            0.75
        );
        
        // Формирование контекста
        String context = buildContext(documents);
        String fullPrompt = buildPrompt(context, request.getMessage());
        
        // Проверка лимита токенов
        ensureTokenLimit(memory, fullPrompt);
        
        // Генерация ответа
        memory.add(UserMessage.from(fullPrompt));
        
        Response<AiMessage> response = chatModel.generate(memory.messages());
        
        memory.add(response.content());
        
        return new ChatResponse(
            response.content().text(),
            documents.stream()
                .map(doc -> doc.getMetadata().get("title"))
                .collect(Collectors.toList())
        );
    }
    
    private ChatMemory getOrCreateMemory(String sessionId) {
        return sessionMemories.computeIfAbsent(
            sessionId,
            id -> MessageWindowChatMemory.withMaxMessages(10)
        );
    }
    
    private String buildContext(List<SearchResult> results) {
        if (results.isEmpty()) {
            return "Релевантной информации в базе знаний не найдено.";
        }
        
        StringBuilder context = new StringBuilder();
        context.append("Информация из базы знаний:
 
");
        
        for (int i = 0; i < results.size(); i++) {
            SearchResult result = results.get(i);
            context.append(String.format("Источник %d (%s):
", 
                i + 1, 
                result.getMetadata().get("title")));
            context.append(result.getContent());
            context.append("
 
---
 
");
        }
        
        return context.toString();
    }
    
    private String buildPrompt(String context, String question) {
        return String.format("""
            %s
            
            Вопрос: %s
            
            Ответь на вопрос, используя предоставленный контекст.
            Если контекста недостаточно - честно скажи об этом.
            При цитировании указывай номер источника.
            """, context, question);
    }
    
    private void ensureTokenLimit(ChatMemory memory, String newPrompt) {
        Tokenizer tokenizer = new OpenAiTokenizer();
        
        int currentTokens = memory.messages().stream()
            .mapToInt(msg -> tokenizer.count(msg.text()))
            .sum();
        
        int newTokens = tokenizer.count(newPrompt);
        
        while (currentTokens + newTokens > maxContextTokens && memory.messages().size() > 2) {
            memory.messages().remove(1); // Удаляем после system message
            memory.messages().remove(1); // Удаляем соответствующий ответ
            
            currentTokens = memory.messages().stream()
                .mapToInt(msg -> tokenizer.count(msg.text()))
                .sum();
        }
    }
    
    public void clearSession(String sessionId) {
        sessionMemories.remove(sessionId);
        logger.info("Очищена сессия: {}", sessionId);
    }
}

@RestController
@RequestMapping("/api/chat")
public class ChatController {
    
    private final ChatService chatService;
    private final ExecutorService executor = Executors.newCachedThreadPool();
    
    @PostMapping
    public ResponseEntity<ChatResponse> chat(@RequestBody ChatRequest request) {
        try {
            ChatResponse response = chatService.chat(request);
            return ResponseEntity.ok(response);
        } catch (Exception e) {
            logger.error("Ошибка обработки запроса", e);
            return ResponseEntity.status(500)
                .body(new ChatResponse("Извините, произошла ошибка", List.of()));
        }
    }
    
    @PostMapping("/stream")
    public SseEmitter stream(@RequestBody ChatRequest request) {
        SseEmitter emitter = new SseEmitter(60_000L);
        
        executor.submit(() -> {
            try {
                chatService.chatStream(request, token -> {
                    try {
                        emitter.send(SseEmitter.event()
                            .name("token")
                            .data(token));
                    } catch (IOException e) {
                        emitter.completeWithError(e);
                    }
                });
                
                emitter.complete();
            } catch (Exception e) {
                emitter.completeWithError(e);
            }
        });
        
        return emitter;
    }
    
    @DeleteMapping("/session/{sessionId}")
    public ResponseEntity<Void> clearSession(@PathVariable String sessionId) {
        chatService.clearSession(sessionId);
        return ResponseEntity.noContent().build();
    }
}

@SpringBootApplication
public class ChatbotApplication {
    
    public static void main(String[] args) {
        SpringApplication.run(ChatbotApplication.class, args);
    }
}

@SpringBootTest
class ChatbotIntegrationTest {
    
    @Autowired
    private DocumentIngestionService ingestionService;
    
    @Autowired
    private ChatService chatService;
    
    @Test
    void testFullWorkflow() {
        // Загрузка тестового документа
        String content = """
            MongoDB Atlas - облачная платформа для работы с MongoDB.
            Она предоставляет автоматическое масштабирование, бэкапы и мониторинг.
            Векторный поиск появился в версии 6.0.11.
            """;
        
        Map<String, Object> metadata = Map.of(
            "author", "test",
            "category", "documentation"
        );
        
        ingestionService.ingestDocument("MongoDB Atlas Overview", content, metadata);
        
        // Даём время на индексацию
        Thread.sleep(2000);
        
        // Тестовый запрос
        ChatRequest request = new ChatRequest(
            "Что такое MongoDB Atlas?",
            "test-session"
        );
        
        ChatResponse response = chatService.chat(request);
        
        assertThat(response.getMessage()).contains("облачная платформа");
        assertThat(response.getSources()).isNotEmpty();
    }
}

@Bean
public Cache<String, Embedding> embeddingCache() {
    return Caffeine.newBuilder()
        .maximumSize(500)
        .maximumWeight(100_000_000)  // ~100MB
        .weigher((String key, Embedding value) -> 
            value.vector().length * 4)  // float = 4 байта
        .expireAfterAccess(Duration.ofHours(2))
        .recordStats()
        .build();
}

@Scheduled(fixedRate = 600000)  // Каждые 10 минут
public void cleanupStaleSessions() {
    long cutoff = System.currentTimeMillis() - TimeUnit.HOURS.toMillis(2);
    
    sessionMemories.entrySet().removeIf(entry -> {
        String sessionId = entry.getKey();
        long lastAccess = sessionLastAccess.getOrDefault(sessionId, 0L);
        
        if (lastAccess < cutoff) {
            logger.debug("Удалена неактивная сессия: {}", sessionId);
            return true;
        }
        return false;
    });
}

@Component
public class RequestQueue {
    
    private final LinkedBlockingQueue<Callable<String>> queue = 
        new LinkedBlockingQueue<>(100);
    
    private final ExecutorService executor = Executors.newFixedThreadPool(5);
    private final Semaphore semaphore = new Semaphore(5);
    
    @PostConstruct
    public void start() {
        for (int i = 0; i < 5; i++) {
            executor.submit(this::processQueue);
        }
    }
    
    private void processQueue() {
        while (!Thread.interrupted()) {
            try {
                Callable<String> task = queue.take();
                semaphore.acquire();
                
                try {
                    task.call();
                } finally {
                    semaphore.release();
                }
                
                Thread.sleep(200);  // Дросселируем запросы
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                logger.error("Queue processing error", e);
            }
        }
    }
    
    public CompletableFuture<String> submit(Callable<String> task) {
        CompletableFuture<String> future = new CompletableFuture<>();
        
        queue.offer(() -> {
            try {
                String result = task.call();
                future.complete(result);
                return result;
            } catch (Exception e) {
                future.completeExceptionally(e);
                throw e;
            }
        });
        
        return future;
    }
}

FROM openjdk:17-slim
 
WORKDIR /app
 
COPY target/chatbot-*.jar app.jar
 
EXPOSE 8080
 
ENV JAVA_OPTS="-Xmx512m -Xms256m"
 
ENTRYPOINT ["sh", "-c", "java $JAVA_OPTS -jar app.jar"]