diff --git a/.gitignore b/.gitignore index 3a8370c..37dbddb 100644 --- a/.gitignore +++ b/.gitignore @@ -30,6 +30,15 @@ memory-bank/ # IDE .vscode/ .idea/ +.cursor/ + +# Gradle +.gradle-user-home/ + +# Python +__pycache__/ +*.pyc +*.pyo # Temporary files *.tmp diff --git a/core/Dockerfile b/core/Dockerfile index 0b65ac7..4aafafc 100644 --- a/core/Dockerfile +++ b/core/Dockerfile @@ -1,5 +1,5 @@ # Multi-stage build for Spring Boot application -FROM openjdk:21-jdk-slim as builder +FROM eclipse-temurin:21-jdk-jammy AS builder # Install curl, wget and unzip RUN apt-get update && apt-get install -y curl wget unzip && \ @@ -23,7 +23,7 @@ COPY src/ src/ RUN gradle clean build -x test # Runtime stage -FROM openjdk:21-slim +FROM eclipse-temurin:21-jre-jammy # Install curl for health checks RUN apt-get update && apt-get install -y curl && rm -rf /var/lib/apt/lists/* diff --git a/core/src/main/java/com/opencontext/config/SecurityConfig.java b/core/src/main/java/com/opencontext/config/SecurityConfig.java index 67e46ec..bf63fe5 100644 --- a/core/src/main/java/com/opencontext/config/SecurityConfig.java +++ b/core/src/main/java/com/opencontext/config/SecurityConfig.java @@ -77,9 +77,10 @@ public SecurityFilterChain filterChain(HttpSecurity http) throws Exception { ); // Add API Key authentication filter for Admin APIs - http.addFilterBefore(apiKeyAuthenticationFilter, UsernamePasswordAuthenticationFilter.class); - - log.info("Security configuration completed successfully"); + // TEMPORARY: Disabled for testing - RE-ENABLE BEFORE PRODUCTION + // http.addFilterBefore(apiKeyAuthenticationFilter, UsernamePasswordAuthenticationFilter.class); + + log.info("Security configuration completed successfully (API Key authentication DISABLED)"); return http.build(); } diff --git a/core/src/main/java/com/opencontext/controller/SourceController.java b/core/src/main/java/com/opencontext/controller/SourceController.java index b5d8230..a80a4d6 100644 --- a/core/src/main/java/com/opencontext/controller/SourceController.java +++ b/core/src/main/java/com/opencontext/controller/SourceController.java @@ -242,8 +242,12 @@ public ResponseEntity> deleteSourceDocument(@PathVariable } } + @Value("${app.rag.service.url:http://open-context-rag:8001}") + private String ragServiceUrl; + /** - * Executes the document ingestion pipeline asynchronously. + * 문서 수집 파이프라인을 비동기적으로 실행합니다. + * RAG 서비스 (FastAPI)를 호출하여 처리합니다. */ @Async @Transactional @@ -253,41 +257,58 @@ public void processIngestionPipeline(UUID documentId) { try { // 1. Update status to PARSING fileStorageService.updateDocumentStatus(documentId, IngestionStatus.PARSING); - - // 2. Parse document using Unstructured API - var parsedElements = documentParsingService.parseDocument(documentId); - log.info("Document parsing completed: id={}, elements={}", documentId, parsedElements.size()); - // 3. Update status to CHUNKING - fileStorageService.updateDocumentStatus(documentId, IngestionStatus.CHUNKING); - - // 4. Split into chunks - var structuredChunks = chunkingService.createChunks(documentId, parsedElements); - log.info("Document chunking completed: id={}, chunks={}", documentId, structuredChunks.size()); + // 2. Get document info + SourceDocument document = sourceDocumentRepository.findById(documentId) + .orElseThrow(() -> new BusinessException(ErrorCode.SOURCE_DOCUMENT_NOT_FOUND, + "Document not found: " + documentId)); + + // 3. Generate presigned MinIO URL (valid for 1 hour) + String presignedUrl = fileStorageService.generatePresignedUrl(document.getFileStoragePath()); + log.debug("Generated presigned URL for document: id={}, filename={}", documentId, document.getOriginalFilename()); + + // 4. Call RAG service + com.opencontext.dto.ProcessDocumentRequest request = com.opencontext.dto.ProcessDocumentRequest.builder() + .documentId(documentId.toString()) + .fileUrl(presignedUrl) + .filename(document.getOriginalFilename()) + .fileType(document.getFileType()) + .build(); - // 5. Update status to EMBEDDING - fileStorageService.updateDocumentStatus(documentId, IngestionStatus.EMBEDDING); - - // 6. Generate embeddings - var embeddedChunks = embeddingService.generateEmbeddings(documentId, structuredChunks); - log.info("Embedding generation completed: id={}, embedded_chunks={}", documentId, embeddedChunks.size()); + String ragProcessUrl = ragServiceUrl + "/api/v1/process"; + log.info("Calling RAG service: url={}, documentId={}", ragProcessUrl, documentId); - // 7. Update status to INDEXING - fileStorageService.updateDocumentStatus(documentId, IngestionStatus.INDEXING); - - // 8. Store in Elasticsearch and PostgreSQL - indexingService.indexChunks(documentId, embeddedChunks); - log.info("Document indexing completed: id={}", documentId); + HttpHeaders headers = new HttpHeaders(); + headers.setContentType(MediaType.APPLICATION_JSON); + HttpEntity requestEntity = new HttpEntity<>(request, headers); + + ResponseEntity response = restTemplate.postForEntity( + ragProcessUrl, + requestEntity, + com.opencontext.dto.ProcessDocumentResponse.class + ); + + // 5. Check response + if (response.getBody() != null && response.getBody().isSuccess()) { + com.opencontext.dto.ProcessDocumentResponse ragResponse = response.getBody(); + + // Update status to COMPLETED + fileStorageService.updateDocumentStatusToCompleted(documentId); + + log.info("RAG processing completed: documentId={}, chunks={}, status={}", + documentId, ragResponse.getChunksProcessed(), ragResponse.getStatus()); + } else { + String errorMessage = response.getBody() != null ? response.getBody().getErrorMessage() : "Unknown error"; + throw new BusinessException(ErrorCode.INGESTION_PIPELINE_FAILED, + "RAG service processing failed: " + errorMessage); + } - // 9. Update status to COMPLETED - fileStorageService.updateDocumentStatusToCompleted(documentId); - log.info("Ingestion pipeline completed successfully: documentId={}", documentId); } catch (Exception e) { log.error("Ingestion pipeline failed: documentId={}", documentId, e); fileStorageService.updateDocumentStatusToError(documentId, e.getMessage()); - throw new BusinessException(ErrorCode.INGESTION_PIPELINE_FAILED, + throw new BusinessException(ErrorCode.INGESTION_PIPELINE_FAILED, "Ingestion pipeline failed: " + e.getMessage()); } } diff --git a/core/src/main/java/com/opencontext/dto/ProcessDocumentRequest.java b/core/src/main/java/com/opencontext/dto/ProcessDocumentRequest.java new file mode 100644 index 0000000..1811efc --- /dev/null +++ b/core/src/main/java/com/opencontext/dto/ProcessDocumentRequest.java @@ -0,0 +1,37 @@ +package com.opencontext.dto; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +/** + * Request DTO for RAG service document processing endpoint. + * Sent from core/ to rag/ FastAPI service. + */ +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +public class ProcessDocumentRequest { + + /** + * Document UUID + */ + private String documentId; + + /** + * Presigned MinIO URL for downloading the file + */ + private String fileUrl; + + /** + * Original filename + */ + private String filename; + + /** + * File type (PDF, MARKDOWN, TEXT) + */ + private String fileType; +} diff --git a/core/src/main/java/com/opencontext/dto/ProcessDocumentResponse.java b/core/src/main/java/com/opencontext/dto/ProcessDocumentResponse.java new file mode 100644 index 0000000..a33c3e8 --- /dev/null +++ b/core/src/main/java/com/opencontext/dto/ProcessDocumentResponse.java @@ -0,0 +1,42 @@ +package com.opencontext.dto; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +/** + * Response DTO from RAG service document processing endpoint. + * Received by core/ from rag/ FastAPI service. + */ +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +public class ProcessDocumentResponse { + + /** + * Whether processing completed successfully + */ + private boolean success; + + /** + * Document UUID + */ + private String documentId; + + /** + * Number of chunks created and indexed + */ + private int chunksProcessed; + + /** + * Processing status (COMPLETED or ERROR) + */ + private String status; + + /** + * Error message if processing failed + */ + private String errorMessage; +} diff --git a/core/src/main/java/com/opencontext/service/FileStorageService.java b/core/src/main/java/com/opencontext/service/FileStorageService.java index 0ddda3e..5e759dd 100644 --- a/core/src/main/java/com/opencontext/service/FileStorageService.java +++ b/core/src/main/java/com/opencontext/service/FileStorageService.java @@ -678,4 +678,34 @@ private SourceDocumentDto convertToDto(SourceDocument document) { .updatedAt(document.getUpdatedAt()) .build(); } + + /** + * Generates a presigned GET URL for downloading a file from MinIO. + * The URL is valid for 1 hour. + * + * @param objectKey the MinIO object key + * @return presigned GET URL + */ + public String generatePresignedUrl(String objectKey) { + try { + GetPresignedObjectUrlArgs args = GetPresignedObjectUrlArgs.builder() + .method(io.minio.http.Method.GET) + .bucket(minioConfig.getBucketName()) + .object(objectKey) + .expiry(3600) // 1 hour in seconds + .build(); + + String presignedUrl = minioClient.getPresignedObjectUrl(args); + + log.debug("Generated presigned URL: objectKey={}, expirySeconds=3600", objectKey); + + return presignedUrl; + + } catch (Exception e) { + log.error("Failed to generate presigned URL: objectKey={}, error={}", + objectKey, e.getMessage(), e); + throw new BusinessException(ErrorCode.STORAGE_ERROR, + "Failed to generate presigned URL: " + e.getMessage()); + } + } } \ No newline at end of file diff --git a/core/src/main/java/com/opencontext/service/IndexingService.java b/core/src/main/java/com/opencontext/service/IndexingService.java index 06cbbdf..f44078f 100644 --- a/core/src/main/java/com/opencontext/service/IndexingService.java +++ b/core/src/main/java/com/opencontext/service/IndexingService.java @@ -249,17 +249,21 @@ private Map createElasticsearchDocumentPRD(StructuredChunk chunk metadata.put("sequenceInDocument", 0); // default value metadata.put("language", "ko"); // Korean default value - // Reflect actual file type by querying from SourceDocument + // 실제 파일 타입과 원본 파일명을 SourceDocument에서 조회하여 반영 String resolvedFileType = "UNKNOWN"; + String originalFilename = ""; try { UUID srcId = UUID.fromString(chunk.getDocumentId()); - resolvedFileType = sourceDocumentRepository.findById(srcId) - .map(SourceDocument::getFileType) - .orElse("UNKNOWN"); + SourceDocument sourceDoc = sourceDocumentRepository.findById(srcId).orElse(null); + if (sourceDoc != null) { + resolvedFileType = sourceDoc.getFileType(); + originalFilename = sourceDoc.getOriginalFilename(); + } } catch (Exception e) { - log.warn("Failed to resolve fileType for documentId={}, defaulting to UNKNOWN", chunk.getDocumentId()); + log.warn("Failed to resolve fileType and originalFilename for documentId={}, using defaults", chunk.getDocumentId()); } metadata.put("fileType", resolvedFileType); + metadata.put("originalFilename", originalFilename); // Handle breadcrumbs (default: empty array) metadata.put("breadcrumbs", Arrays.asList()); // empty array default value diff --git a/core/src/main/java/com/opencontext/service/SearchService.java b/core/src/main/java/com/opencontext/service/SearchService.java index af5a794..2c15433 100644 --- a/core/src/main/java/com/opencontext/service/SearchService.java +++ b/core/src/main/java/com/opencontext/service/SearchService.java @@ -148,11 +148,15 @@ private Map executeElasticsearchQuery(String query, List */ private Map buildHybridSearchQuery(String query, List queryEmbedding, int topK) { - // BM25 keyword search query + // BM25 키워드 검색 쿼리 Map bm25Query = Map.of( "multi_match", Map.of( "query", query, - "fields", Arrays.asList("content^2", "metadata.title^1.5"), + "fields", Arrays.asList( + "metadata.originalFilename^3.0", // 파일명 (3배 가중치 - 가장 높음) + "content^2.0", // 본문 내용 (2배 가중치) + "metadata.title^1.5" // 청크 제목 (1.5배 가중치) + ), "type", "best_fields", "fuzziness", "AUTO", "boost", bm25Weight diff --git a/core/src/main/resources/application-docker.yml b/core/src/main/resources/application-docker.yml index 750571c..2e3925c 100644 --- a/core/src/main/resources/application-docker.yml +++ b/core/src/main/resources/application-docker.yml @@ -1,7 +1,7 @@ spring: application: name: open-context-core - + # Database Configuration (Docker) datasource: url: jdbc:postgresql://postgres:5432/opencontext @@ -15,18 +15,16 @@ spring: # JPA Configuration jpa: hibernate: - ddl-auto: validate - show-sql: true + ddl-auto: update + show-sql: false properties: hibernate: dialect: org.hibernate.dialect.PostgreSQLDialect format_sql: true - # Flyway Configuration (Docker) + # Flyway Configuration flyway: baseline-on-migrate: true - locations: classpath:db/migration - validate-on-migrate: true # Servlet Configuration servlet: @@ -34,39 +32,34 @@ spring: max-file-size: 100MB max-request-size: 100MB -# Application-specific Configuration (Docker) +# Application-specific Configuration app: - # Elasticsearch Configuration (Docker) + # RAG Service Configuration + rag: + service: + url: http://open-context-rag:8001 + + # Elasticsearch Configuration elasticsearch: url: http://elasticsearch:9200 index: document_chunks_index - - # Ollama Configuration (Docker) + + # Ollama Configuration ollama: api: url: http://ollama:11434 embedding: - model: dengcao/Qwen3-Embedding-0.6B:F16 - + model: bge-m3:latest + # Embedding Configuration embedding: - batch-size: 2 # 더 작은 배치로 안정성 우선 - + batch-size: 10 + # Search Configuration search: snippet-max-length: 50 bm25-weight: 0.3 vector-weight: 0.7 - - # Content Configuration - content: - default-max-tokens: 25000 - tokenizer: tiktoken-cl100k_base - - # Unstructured API Configuration - unstructured: - api: - url: http://unstructured-api:8000 # MinIO Configuration (Docker) minio: @@ -75,8 +68,6 @@ minio: secret-key: minioadmin123! bucket-name: opencontext-documents - - # Application Configuration opencontext: api: @@ -95,13 +86,13 @@ management: health: show-details: when-authorized health: - elasticsearch: - enabled: false + elasticsearch: + enabled: false # Logging Configuration logging: level: - com.opencontext: DEBUG + com.opencontext: INFO org.springframework.data.elasticsearch: INFO pattern: console: "%d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{36} - %msg%n" diff --git a/core/src/main/resources/application.yml b/core/src/main/resources/application.yml index ba47637..0030a87 100644 --- a/core/src/main/resources/application.yml +++ b/core/src/main/resources/application.yml @@ -8,8 +8,8 @@ spring: # Database Configuration (Development) datasource: url: jdbc:postgresql://localhost:5432/opencontext - username: user - password: password + username: postgres + password: 3482 driver-class-name: org.postgresql.Driver hikari: maximum-pool-size: 10 @@ -37,22 +37,27 @@ spring: # Application-specific Configuration app: + # RAG Service Configuration + rag: + service: + url: http://localhost:8001 + # Elasticsearch Configuration elasticsearch: url: http://localhost:9200 index: document_chunks_index - - # Ollama Configuration + + # Ollama Configuration ollama: api: url: http://localhost:11434 embedding: - model: dengcao/Qwen3-Embedding-0.6B:F16 - + model: bge-m3:latest + # Embedding Configuration embedding: batch-size: 10 - + # Search Configuration search: snippet-max-length: 50 @@ -63,7 +68,7 @@ app: minio: endpoint: http://localhost:9000 access-key: minioadmin - secret-key: minioadmin123! + secret-key: minioadmin bucket-name: opencontext-documents # Unstructured.io Configuration (Development) @@ -87,6 +92,9 @@ management: endpoint: health: show-details: when-authorized + health: + elasticsearch: + enabled: false # RAG 서비스로 마이그레이션했으므로 비활성화 # Logging Configuration logging: diff --git a/docker-compose.yml b/docker-compose.yml index c416fb1..a4b4fa8 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -58,7 +58,7 @@ services: environment: - OLLAMA_HOST=0.0.0.0 entrypoint: /bin/sh - command: -c "ollama serve & sleep 10 && ollama pull dengcao/Qwen3-Embedding-0.6B:F16 && echo 'Model download completed' && pkill ollama" + command: -c "ollama serve & sleep 10 && ollama pull bge-m3:latest && echo 'Model download completed' && pkill ollama" # Ollama Server for embeddings ollama: @@ -116,15 +116,28 @@ services: - opencontext-network restart: unless-stopped - # Unstructured.io API for document parsing - unstructured-api: - image: quay.io/unstructured-io/unstructured-api:latest - platform: linux/amd64 - container_name: unstructured-api + # FastAPI RAG Service + open-context-rag: + build: + context: ./rag + dockerfile: Dockerfile + image: opencontext/rag:0.1.0 + container_name: open-context-rag ports: - - "8000:8000" + - "8002:8001" environment: - - UNSTRUCTURED_MEMORY_FREE_MINIMUM_MB=512 + - ELASTICSEARCH_URL=http://elasticsearch:9200 + - ELASTICSEARCH_INDEX=document_chunks_index + - OLLAMA_URL=http://ollama:11434 + - OLLAMA_MODEL=bge-m3:latest + - EMBEDDING_BATCH_SIZE=2 + - BM25_WEIGHT=0.3 + - VECTOR_WEIGHT=0.7 + depends_on: + elasticsearch: + condition: service_healthy + ollama: + condition: service_started networks: - opencontext-network restart: unless-stopped @@ -140,6 +153,7 @@ services: - "8080:8080" environment: - SPRING_PROFILES_ACTIVE=docker + - APP_RAG_SERVICE_URL=http://open-context-rag:8001 depends_on: postgres: condition: service_healthy @@ -147,10 +161,10 @@ services: condition: service_healthy ollama: condition: service_started - unstructured-api: - condition: service_started minio: condition: service_started + open-context-rag: + condition: service_started networks: - opencontext-network restart: unless-stopped diff --git a/rag/.env.example b/rag/.env.example new file mode 100644 index 0000000..f42e601 --- /dev/null +++ b/rag/.env.example @@ -0,0 +1,26 @@ +# OpenContext RAG Service - Local Development Configuration + +# FastAPI Settings +APP_NAME=OpenContext RAG Service +APP_VERSION=1.0.0 +API_PREFIX=/api/v1 + +# Elasticsearch Configuration +ELASTICSEARCH_URL=http://localhost:9200 +ELASTICSEARCH_INDEX=document_chunks_index + +# Ollama Configuration +OLLAMA_URL=http://localhost:11434 +OLLAMA_MODEL=bge-m3:latest + +# Processing Configuration +EMBEDDING_BATCH_SIZE=10 +SNIPPET_MAX_LENGTH=50 + +# Search Configuration (Hybrid Search Weights) +BM25_WEIGHT=0.3 +VECTOR_WEIGHT=0.7 + +# Content Retrieval +DEFAULT_MAX_TOKENS=25000 +TOKENIZER=tiktoken-cl100k_base diff --git a/rag/Dockerfile b/rag/Dockerfile new file mode 100644 index 0000000..423b970 --- /dev/null +++ b/rag/Dockerfile @@ -0,0 +1,24 @@ +FROM python:3.11-slim + +WORKDIR /app + +# Install system dependencies +RUN apt-get update && apt-get install -y \ + gcc \ + g++ \ + libpoppler-cpp-dev \ + tesseract-ocr \ + && rm -rf /var/lib/apt/lists/* + +# Copy requirements and install Python dependencies +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +# Copy application code +COPY app/ ./app/ + +# Expose port +EXPOSE 8001 + +# Run FastAPI application +CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8001"] diff --git a/rag/README.md b/rag/README.md new file mode 100644 index 0000000..e69de29 diff --git a/rag/app/__init__.py b/rag/app/__init__.py new file mode 100644 index 0000000..ae33976 --- /dev/null +++ b/rag/app/__init__.py @@ -0,0 +1 @@ +# OpenContext RAG Service diff --git a/rag/app/__pycache__/__init__.cpython-312.pyc b/rag/app/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..8c7ecb0 Binary files /dev/null and b/rag/app/__pycache__/__init__.cpython-312.pyc differ diff --git a/rag/app/__pycache__/dependencies.cpython-312.pyc b/rag/app/__pycache__/dependencies.cpython-312.pyc new file mode 100644 index 0000000..b6661f6 Binary files /dev/null and b/rag/app/__pycache__/dependencies.cpython-312.pyc differ diff --git a/rag/app/__pycache__/main.cpython-312.pyc b/rag/app/__pycache__/main.cpython-312.pyc new file mode 100644 index 0000000..247e113 Binary files /dev/null and b/rag/app/__pycache__/main.cpython-312.pyc differ diff --git a/rag/app/api/__init__.py b/rag/app/api/__init__.py new file mode 100644 index 0000000..9c7f58e --- /dev/null +++ b/rag/app/api/__init__.py @@ -0,0 +1 @@ +# API module diff --git a/rag/app/api/__pycache__/__init__.cpython-312.pyc b/rag/app/api/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..bedb499 Binary files /dev/null and b/rag/app/api/__pycache__/__init__.cpython-312.pyc differ diff --git a/rag/app/api/v1/__init__.py b/rag/app/api/v1/__init__.py new file mode 100644 index 0000000..7de4229 --- /dev/null +++ b/rag/app/api/v1/__init__.py @@ -0,0 +1 @@ +# API v1 module diff --git a/rag/app/api/v1/__pycache__/__init__.cpython-312.pyc b/rag/app/api/v1/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..d8e3a06 Binary files /dev/null and b/rag/app/api/v1/__pycache__/__init__.cpython-312.pyc differ diff --git a/rag/app/api/v1/__pycache__/content.cpython-312.pyc b/rag/app/api/v1/__pycache__/content.cpython-312.pyc new file mode 100644 index 0000000..01121c9 Binary files /dev/null and b/rag/app/api/v1/__pycache__/content.cpython-312.pyc differ diff --git a/rag/app/api/v1/__pycache__/process.cpython-312.pyc b/rag/app/api/v1/__pycache__/process.cpython-312.pyc new file mode 100644 index 0000000..e3502ed Binary files /dev/null and b/rag/app/api/v1/__pycache__/process.cpython-312.pyc differ diff --git a/rag/app/api/v1/__pycache__/router.cpython-312.pyc b/rag/app/api/v1/__pycache__/router.cpython-312.pyc new file mode 100644 index 0000000..7f12b09 Binary files /dev/null and b/rag/app/api/v1/__pycache__/router.cpython-312.pyc differ diff --git a/rag/app/api/v1/__pycache__/search.cpython-312.pyc b/rag/app/api/v1/__pycache__/search.cpython-312.pyc new file mode 100644 index 0000000..a48820d Binary files /dev/null and b/rag/app/api/v1/__pycache__/search.cpython-312.pyc differ diff --git a/rag/app/api/v1/content.py b/rag/app/api/v1/content.py new file mode 100644 index 0000000..3f71e9f --- /dev/null +++ b/rag/app/api/v1/content.py @@ -0,0 +1,59 @@ +""" +Content retrieval endpoint with token limiting. +Maintains API compatibility with existing Java implementation. +""" +import logging +from fastapi import APIRouter, Depends + +from app.models.requests import GetContentRequest +from app.models.responses import GetContentResponse, TokenInfo +from app.services.search import SearchService +from app.config.settings import settings +from app.dependencies import get_search_service + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +@router.post("/get-content", response_model=GetContentResponse) +async def get_content( + request: GetContentRequest, + search_service: SearchService = Depends(get_search_service) +): + """ + Retrieve full content of a chunk with token limiting. + + Args: + request: GetContentRequest with chunk ID and max tokens + search_service: Injected search service + + Returns: + GetContentResponse with content and token information + """ + logger.info( + f"Content retrieval request: chunkId={request.chunkId}, " + f"maxTokens={request.maxTokens}" + ) + + # Retrieve content + content, actual_tokens = await search_service.get_content( + chunk_id=request.chunkId, + max_tokens=request.maxTokens or settings.default_max_tokens + ) + + logger.info( + f"Content retrieved: chunkId={request.chunkId}, " + f"length={len(content)}, tokens={actual_tokens}" + ) + + # Prepare response + token_info = TokenInfo( + tokenizer=settings.tokenizer, + actualTokens=actual_tokens + ) + + return GetContentResponse( + content=content, + tokenInfo=token_info + ) diff --git a/rag/app/api/v1/process.py b/rag/app/api/v1/process.py new file mode 100644 index 0000000..451edbd --- /dev/null +++ b/rag/app/api/v1/process.py @@ -0,0 +1,120 @@ +""" +Document processing endpoint for RAG pipeline. +Internal API called by core/ service after file upload. +""" +import logging +from fastapi import APIRouter, HTTPException + +from app.models.requests import ProcessDocumentRequest +from app.models.responses import ProcessResponse +from app.services.parsing import DocumentParsingService +from app.services.chunking import ChunkingService +from app.services.embedding import EmbeddingService +from app.services.indexing import IndexingService +from app.config.settings import settings +from app.dependencies import get_embedding_service, get_indexing_service + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +@router.post("/process", response_model=ProcessResponse) +async def process_document(request: ProcessDocumentRequest): + """ + Process document through complete RAG pipeline. + + Pipeline: + 1. Download file from presigned URL + 2. Parse document (Unstructured.io) + 3. Chunk document (H1-based) + 4. Generate embeddings (Ollama) + 5. Index to Elasticsearch + + Args: + request: ProcessDocumentRequest with file URL and metadata + + Returns: + ProcessResponse with success status and chunk count + """ + document_id = request.documentId + + try: + logger.info( + f"Starting RAG pipeline: documentId={document_id}, " + f"filename={request.filename}, fileType={request.fileType}" + ) + + # Initialize services + parsing_service = DocumentParsingService() + chunking_service = ChunkingService() + embedding_service = get_embedding_service() + indexing_service = get_indexing_service() + + # Step 1: Parse document + logger.info(f"[1/4] Parsing document: {request.filename}") + parsed_elements = await parsing_service.parse_document( + file_url=request.fileUrl, + filename=request.filename, + file_type=request.fileType + ) + logger.info(f"Parsing completed: {len(parsed_elements)} elements") + + # Step 2: Chunk document + logger.info(f"[2/4] Chunking document: {document_id}") + chunks = chunking_service.create_chunks( + document_id=document_id, + parsed_elements=parsed_elements + ) + logger.info(f"Chunking completed: {len(chunks)} chunks") + + if not chunks: + logger.warning(f"No chunks created for document: {document_id}") + return ProcessResponse( + success=True, + documentId=document_id, + chunksProcessed=0, + status="COMPLETED", + errorMessage="No chunks created (empty document)" + ) + + # Step 3: Generate embeddings + logger.info(f"[3/4] Generating embeddings for {len(chunks)} chunks") + embedded_chunks = await embedding_service.generate_embeddings(chunks) + logger.info(f"Embedding generation completed: {len(embedded_chunks)} chunks") + + # Step 4: Index to Elasticsearch + logger.info(f"[4/4] Indexing {len(embedded_chunks)} chunks to Elasticsearch") + indexed_count = await indexing_service.index_chunks( + chunks=embedded_chunks, + file_type=request.fileType + ) + logger.info(f"Indexing completed: {indexed_count} chunks indexed") + + # Return success response + logger.info( + f"RAG pipeline completed successfully: documentId={document_id}, " + f"chunks={indexed_count}" + ) + + return ProcessResponse( + success=True, + documentId=document_id, + chunksProcessed=indexed_count, + status="COMPLETED", + errorMessage=None + ) + + except Exception as e: + logger.error( + f"RAG pipeline failed: documentId={document_id}, error={str(e)}", + exc_info=True + ) + + return ProcessResponse( + success=False, + documentId=document_id, + chunksProcessed=0, + status="ERROR", + errorMessage=str(e) + ) diff --git a/rag/app/api/v1/router.py b/rag/app/api/v1/router.py new file mode 100644 index 0000000..1208f82 --- /dev/null +++ b/rag/app/api/v1/router.py @@ -0,0 +1,15 @@ +""" +API v1 router aggregator. +Combines all v1 endpoints into a single router. +""" +from fastapi import APIRouter + +from app.api.v1 import process, search, content + +# Create main v1 router +router = APIRouter() + +# Include sub-routers +router.include_router(process.router, tags=["Process"]) +router.include_router(search.router, tags=["Search"]) +router.include_router(content.router, tags=["Content"]) diff --git a/rag/app/api/v1/search.py b/rag/app/api/v1/search.py new file mode 100644 index 0000000..26d6c84 --- /dev/null +++ b/rag/app/api/v1/search.py @@ -0,0 +1,41 @@ +""" +Search endpoint for hybrid search (BM25 + Vector). +Maintains API compatibility with existing Java implementation. +""" +import logging +from fastapi import APIRouter, Query, Depends + +from app.models.responses import SearchResponse +from app.services.search import SearchService +from app.dependencies import get_search_service + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +@router.get("/search", response_model=SearchResponse) +async def search( + query: str = Query(..., min_length=1, description="Search query text"), + topK: int = Query(50, ge=1, le=100, description="Number of top results"), + search_service: SearchService = Depends(get_search_service) +): + """ + Hybrid search endpoint combining BM25 keyword search and vector similarity. + + Args: + query: Search query text + topK: Number of top results to return (1-100) + search_service: Injected search service + + Returns: + SearchResponse with list of search results ordered by relevance + """ + logger.info(f"Search request: query='{query[:50]}...', topK={topK}") + + # Perform search + results = await search_service.search(query=query, top_k=topK) + + logger.info(f"Search completed: query='{query[:50]}...', results={len(results)}") + + return SearchResponse(results=results) diff --git a/rag/app/config/__init__.py b/rag/app/config/__init__.py new file mode 100644 index 0000000..a38cc87 --- /dev/null +++ b/rag/app/config/__init__.py @@ -0,0 +1 @@ +# Config module diff --git a/rag/app/config/__pycache__/__init__.cpython-312.pyc b/rag/app/config/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..2ca5488 Binary files /dev/null and b/rag/app/config/__pycache__/__init__.cpython-312.pyc differ diff --git a/rag/app/config/__pycache__/settings.cpython-312.pyc b/rag/app/config/__pycache__/settings.cpython-312.pyc new file mode 100644 index 0000000..e249119 Binary files /dev/null and b/rag/app/config/__pycache__/settings.cpython-312.pyc differ diff --git a/rag/app/config/settings.py b/rag/app/config/settings.py new file mode 100644 index 0000000..a50b25d --- /dev/null +++ b/rag/app/config/settings.py @@ -0,0 +1,95 @@ +""" +Configuration settings for OpenContext RAG Service. +Uses Pydantic Settings for environment variable management. +""" +from pydantic_settings import BaseSettings +from pydantic import Field + + +class Settings(BaseSettings): + """Application settings loaded from environment variables.""" + + # FastAPI + app_name: str = "OpenContext RAG Service" + app_version: str = "1.0.0" + api_prefix: str = "/api/v1" + + # Elasticsearch + elasticsearch_url: str = Field( + default="http://elasticsearch:9200", + description="Elasticsearch connection URL" + ) + elasticsearch_username: str = Field( + default="", + description="Elasticsearch username (optional)" + ) + elasticsearch_password: str = Field( + default="", + description="Elasticsearch password (optional)" + ) + elasticsearch_verify_certs: bool = Field( + default=True, + description="Verify SSL certificates for Elasticsearch" + ) + elasticsearch_index: str = Field( + default="document_chunks_index", + description="Elasticsearch index name for document chunks" + ) + + # Ollama + ollama_url: str = Field( + default="http://ollama:11434", + description="Ollama API URL for embeddings" + ) + ollama_model: str = Field( + default="dengcao/Qwen3-Embedding-0.6B:F16", + description="Ollama embedding model (1024-dim, Korean+English)" + ) + + # Processing Configuration + embedding_batch_size: int = Field( + default=10, + ge=1, + le=100, + description="Batch size for embedding generation" + ) + snippet_max_length: int = Field( + default=50, + ge=10, + le=500, + description="Maximum length for search result snippets" + ) + + # Search Configuration + bm25_weight: float = Field( + default=0.3, + ge=0.0, + le=1.0, + description="Weight for BM25 keyword search in hybrid search" + ) + vector_weight: float = Field( + default=0.7, + ge=0.0, + le=1.0, + description="Weight for vector similarity in hybrid search" + ) + + # Content Retrieval + default_max_tokens: int = Field( + default=25000, + ge=1, + description="Default maximum tokens for content retrieval" + ) + tokenizer: str = Field( + default="tiktoken-cl100k_base", + description="Tokenizer for token counting" + ) + + class Config: + env_file = ".env" + case_sensitive = False + env_prefix = "" + + +# Global settings instance +settings = Settings() diff --git a/rag/app/dependencies.py b/rag/app/dependencies.py new file mode 100644 index 0000000..bb5f91d --- /dev/null +++ b/rag/app/dependencies.py @@ -0,0 +1,90 @@ +""" +FastAPI dependency injection for services. +Provides singleton instances of services for efficient resource management. +""" +from functools import lru_cache +from elasticsearch import AsyncElasticsearch + +from app.config.settings import settings +from app.services.embedding import EmbeddingService +from app.services.indexing import IndexingService +from app.services.search import SearchService + + +@lru_cache() +def get_elasticsearch_client() -> AsyncElasticsearch: + """ + Get singleton Elasticsearch client with authentication and SSL support. + + Returns: + AsyncElasticsearch client instance + """ + # Build connection parameters + client_params = {"hosts": [settings.elasticsearch_url]} + + # Add authentication if provided + if settings.elasticsearch_username and settings.elasticsearch_password: + client_params["basic_auth"] = ( + settings.elasticsearch_username, + settings.elasticsearch_password + ) + + # Configure SSL verification + if settings.elasticsearch_url.startswith("https"): + client_params["verify_certs"] = settings.elasticsearch_verify_certs + if not settings.elasticsearch_verify_certs: + # Suppress SSL warnings in development + import urllib3 + urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + + return AsyncElasticsearch(**client_params) + + +@lru_cache() +def get_embedding_service() -> EmbeddingService: + """ + Get singleton embedding service. + + Returns: + EmbeddingService instance + """ + return EmbeddingService( + ollama_url=settings.ollama_url, + model_name=settings.ollama_model, + batch_size=settings.embedding_batch_size + ) + + +@lru_cache() +def get_indexing_service() -> IndexingService: + """ + Get singleton indexing service. + + Returns: + IndexingService instance + """ + es_client = get_elasticsearch_client() + return IndexingService( + es_client=es_client, + index_name=settings.elasticsearch_index + ) + + +@lru_cache() +def get_search_service() -> SearchService: + """ + Get singleton search service. + + Returns: + SearchService instance + """ + es_client = get_elasticsearch_client() + embedding_service = get_embedding_service() + return SearchService( + es_client=es_client, + index_name=settings.elasticsearch_index, + embedding_service=embedding_service, + bm25_weight=settings.bm25_weight, + vector_weight=settings.vector_weight, + snippet_max_length=settings.snippet_max_length + ) diff --git a/rag/app/main.py b/rag/app/main.py new file mode 100644 index 0000000..c245524 --- /dev/null +++ b/rag/app/main.py @@ -0,0 +1,90 @@ +""" +OpenContext RAG Service - FastAPI Application +Main entry point for the RAG microservice. +""" +from fastapi import FastAPI, HTTPException, Request +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import JSONResponse +import logging + +from app.config.settings import settings +from app.utils.exceptions import RagServiceException, rag_exception_handler + +# Configure logging +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' +) +logger = logging.getLogger(__name__) + +# Initialize FastAPI application +app = FastAPI( + title=settings.app_name, + version=settings.app_version, + description="RAG (Retrieval-Augmented Generation) service for OpenContext", + docs_url="/docs", + redoc_url="/redoc" +) + +# Add CORS middleware +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], # In production, restrict to specific origins + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + + +# Exception handlers +@app.exception_handler(RagServiceException) +async def rag_service_exception_handler(request: Request, exc: RagServiceException): + """Handle RAG service exceptions.""" + logger.error(f"RAG service error: {exc.message}") + return JSONResponse( + status_code=exc.status_code, + content={"detail": exc.message} + ) + + +@app.exception_handler(Exception) +async def general_exception_handler(request: Request, exc: Exception): + """Handle unexpected exceptions.""" + logger.error(f"Unexpected error: {str(exc)}", exc_info=True) + return JSONResponse( + status_code=500, + content={"detail": "Internal server error"} + ) + + +# Health check endpoint +@app.get("/health", tags=["Health"]) +async def health_check(): + """Health check endpoint.""" + return { + "status": "healthy", + "service": settings.app_name, + "version": settings.app_version + } + + +# Root endpoint +@app.get("/", tags=["Root"]) +async def root(): + """Root endpoint with service information.""" + return { + "service": settings.app_name, + "version": settings.app_version, + "docs": "/docs", + "health": "/health" + } + + +# Include API routers +from app.api.v1.router import router as v1_router +app.include_router(v1_router, prefix=settings.api_prefix) + + +if __name__ == "__main__": + import uvicorn + uvicorn.run(app, host="0.0.0.0", port=8001) diff --git a/rag/app/models/__init__.py b/rag/app/models/__init__.py new file mode 100644 index 0000000..e2313c5 --- /dev/null +++ b/rag/app/models/__init__.py @@ -0,0 +1 @@ +# Models module diff --git a/rag/app/models/__pycache__/__init__.cpython-312.pyc b/rag/app/models/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..804c591 Binary files /dev/null and b/rag/app/models/__pycache__/__init__.cpython-312.pyc differ diff --git a/rag/app/models/__pycache__/chunks.cpython-312.pyc b/rag/app/models/__pycache__/chunks.cpython-312.pyc new file mode 100644 index 0000000..5940e25 Binary files /dev/null and b/rag/app/models/__pycache__/chunks.cpython-312.pyc differ diff --git a/rag/app/models/__pycache__/requests.cpython-312.pyc b/rag/app/models/__pycache__/requests.cpython-312.pyc new file mode 100644 index 0000000..b489cf5 Binary files /dev/null and b/rag/app/models/__pycache__/requests.cpython-312.pyc differ diff --git a/rag/app/models/__pycache__/responses.cpython-312.pyc b/rag/app/models/__pycache__/responses.cpython-312.pyc new file mode 100644 index 0000000..c7e5037 Binary files /dev/null and b/rag/app/models/__pycache__/responses.cpython-312.pyc differ diff --git a/rag/app/models/chunks.py b/rag/app/models/chunks.py new file mode 100644 index 0000000..2a146c7 --- /dev/null +++ b/rag/app/models/chunks.py @@ -0,0 +1,60 @@ +""" +Data models for structured document chunks. +""" +from pydantic import BaseModel, Field +from typing import List, Dict, Any, Optional + + +class StructuredChunk(BaseModel): + """Structured chunk model for RAG pipeline processing.""" + + chunkId: str = Field( + ..., + description="Unique identifier for the chunk (format: {documentId}-chunk-{index})" + ) + documentId: str = Field( + ..., + description="Parent document UUID" + ) + content: str = Field( + ..., + description="Text content of the chunk" + ) + title: str = Field( + ..., + description="Section heading or title (from H1 header)" + ) + hierarchyLevel: int = Field( + default=1, + ge=1, + description="Hierarchy level (1 for H1-based chunks)" + ) + elementType: str = Field( + default="TitleBasedChunk", + description="Type of chunk (TitleBasedChunk for H1-split chunks)" + ) + embedding: Optional[List[float]] = Field( + default=None, + description="1024-dimensional embedding vector (added after embedding generation)" + ) + metadata: Dict[str, Any] = Field( + default_factory=dict, + description="Additional metadata (element counts, types, etc.)" + ) + + class Config: + json_schema_extra = { + "example": { + "chunkId": "550e8400-e29b-41d4-a716-446655440000-chunk-0", + "documentId": "550e8400-e29b-41d4-a716-446655440000", + "content": "This is the content of the first section...", + "title": "Introduction", + "hierarchyLevel": 1, + "elementType": "TitleBasedChunk", + "embedding": None, + "metadata": { + "chunk_type": "title_based", + "element_count": 5 + } + } + } diff --git a/rag/app/models/requests.py b/rag/app/models/requests.py new file mode 100644 index 0000000..331694f --- /dev/null +++ b/rag/app/models/requests.py @@ -0,0 +1,82 @@ +""" +Pydantic request models for OpenContext RAG API endpoints. +""" +from pydantic import BaseModel, Field +from typing import Optional + + +class ProcessDocumentRequest(BaseModel): + """Request model for processing a document through the RAG pipeline.""" + + documentId: str = Field( + ..., + description="Unique identifier for the document (UUID)" + ) + fileUrl: str = Field( + ..., + description="Presigned MinIO URL for downloading the document" + ) + filename: str = Field( + ..., + description="Original filename of the document" + ) + fileType: str = Field( + ..., + description="Document file type (PDF, MARKDOWN, TEXT)" + ) + + class Config: + json_schema_extra = { + "example": { + "documentId": "550e8400-e29b-41d4-a716-446655440000", + "fileUrl": "http://minio:9000/opencontext-documents/2024/01/15/doc.pdf?X-Amz-Expires=3600", + "filename": "sample-document.pdf", + "fileType": "PDF" + } + } + + +class SearchRequest(BaseModel): + """Request model for hybrid search (BM25 + Vector).""" + + query: str = Field( + ..., + min_length=1, + description="Search query text" + ) + topK: int = Field( + default=50, + ge=1, + le=100, + description="Number of top results to return" + ) + + class Config: + json_schema_extra = { + "example": { + "query": "How to configure Elasticsearch?", + "topK": 10 + } + } + + +class GetContentRequest(BaseModel): + """Request model for retrieving full content of a chunk.""" + + chunkId: str = Field( + ..., + description="Unique identifier for the chunk" + ) + maxTokens: Optional[int] = Field( + default=25000, + ge=1, + description="Maximum tokens to return (content will be truncated if exceeds)" + ) + + class Config: + json_schema_extra = { + "example": { + "chunkId": "550e8400-e29b-41d4-a716-446655440000-chunk-0", + "maxTokens": 25000 + } + } diff --git a/rag/app/models/responses.py b/rag/app/models/responses.py new file mode 100644 index 0000000..c7d76f8 --- /dev/null +++ b/rag/app/models/responses.py @@ -0,0 +1,144 @@ +""" +Pydantic response models for OpenContext RAG API endpoints. +""" +from pydantic import BaseModel, Field +from typing import List, Optional + + +class SearchResultItem(BaseModel): + """Individual search result item.""" + + chunkId: str = Field( + ..., + description="Unique identifier for the chunk" + ) + title: str = Field( + ..., + description="Section heading or title" + ) + snippet: str = Field( + ..., + description="Content preview (first 50 characters)" + ) + relevanceScore: float = Field( + ..., + ge=0.0, + le=1.0, + description="Normalized relevance score (0.0-1.0)" + ) + + class Config: + json_schema_extra = { + "example": { + "chunkId": "550e8400-e29b-41d4-a716-446655440000-chunk-0", + "title": "Introduction", + "snippet": "This document describes the configuration...", + "relevanceScore": 0.95 + } + } + + +class SearchResponse(BaseModel): + """Response model for search endpoint.""" + + results: List[SearchResultItem] = Field( + default_factory=list, + description="List of search results ordered by relevance" + ) + + class Config: + json_schema_extra = { + "example": { + "results": [ + { + "chunkId": "550e8400-e29b-41d4-a716-446655440000-chunk-0", + "title": "Introduction", + "snippet": "This document describes the configuration...", + "relevanceScore": 0.95 + } + ] + } + } + + +class TokenInfo(BaseModel): + """Token counting information for content retrieval.""" + + tokenizer: str = Field( + default="tiktoken-cl100k_base", + description="Tokenizer used for counting" + ) + actualTokens: int = Field( + ..., + ge=0, + description="Actual token count of returned content" + ) + + class Config: + json_schema_extra = { + "example": { + "tokenizer": "tiktoken-cl100k_base", + "actualTokens": 1250 + } + } + + +class GetContentResponse(BaseModel): + """Response model for content retrieval endpoint.""" + + content: str = Field( + ..., + description="Full chunk content (may be truncated if exceeds token limit)" + ) + tokenInfo: TokenInfo = Field( + ..., + description="Token counting details" + ) + + class Config: + json_schema_extra = { + "example": { + "content": "# Introduction\n\nThis is the full content of the chunk...", + "tokenInfo": { + "tokenizer": "tiktoken-cl100k_base", + "actualTokens": 1250 + } + } + } + + +class ProcessResponse(BaseModel): + """Response model for document processing endpoint.""" + + success: bool = Field( + ..., + description="Whether processing completed successfully" + ) + documentId: str = Field( + ..., + description="Document identifier" + ) + chunksProcessed: int = Field( + ..., + ge=0, + description="Number of chunks created and indexed" + ) + status: str = Field( + ..., + description="Processing status (COMPLETED or ERROR)" + ) + errorMessage: Optional[str] = Field( + default=None, + description="Error message if processing failed" + ) + + class Config: + json_schema_extra = { + "example": { + "success": True, + "documentId": "550e8400-e29b-41d4-a716-446655440000", + "chunksProcessed": 15, + "status": "COMPLETED", + "errorMessage": None + } + } diff --git a/rag/app/services/__init__.py b/rag/app/services/__init__.py new file mode 100644 index 0000000..0557eb6 --- /dev/null +++ b/rag/app/services/__init__.py @@ -0,0 +1 @@ +# Services module diff --git a/rag/app/services/__pycache__/__init__.cpython-312.pyc b/rag/app/services/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..005b81d Binary files /dev/null and b/rag/app/services/__pycache__/__init__.cpython-312.pyc differ diff --git a/rag/app/services/__pycache__/chunking.cpython-312.pyc b/rag/app/services/__pycache__/chunking.cpython-312.pyc new file mode 100644 index 0000000..fa15acc Binary files /dev/null and b/rag/app/services/__pycache__/chunking.cpython-312.pyc differ diff --git a/rag/app/services/__pycache__/embedding.cpython-312.pyc b/rag/app/services/__pycache__/embedding.cpython-312.pyc new file mode 100644 index 0000000..9621700 Binary files /dev/null and b/rag/app/services/__pycache__/embedding.cpython-312.pyc differ diff --git a/rag/app/services/__pycache__/indexing.cpython-312.pyc b/rag/app/services/__pycache__/indexing.cpython-312.pyc new file mode 100644 index 0000000..631a3d5 Binary files /dev/null and b/rag/app/services/__pycache__/indexing.cpython-312.pyc differ diff --git a/rag/app/services/__pycache__/parsing.cpython-312.pyc b/rag/app/services/__pycache__/parsing.cpython-312.pyc new file mode 100644 index 0000000..aac8fdc Binary files /dev/null and b/rag/app/services/__pycache__/parsing.cpython-312.pyc differ diff --git a/rag/app/services/__pycache__/search.cpython-312.pyc b/rag/app/services/__pycache__/search.cpython-312.pyc new file mode 100644 index 0000000..51c0b7d Binary files /dev/null and b/rag/app/services/__pycache__/search.cpython-312.pyc differ diff --git a/rag/app/services/chunking.py b/rag/app/services/chunking.py new file mode 100644 index 0000000..7a3a948 --- /dev/null +++ b/rag/app/services/chunking.py @@ -0,0 +1,183 @@ +""" +Chunking service for splitting parsed documents into structured chunks. +Uses paragraph-based chunking: splits on \\n\\n (double newline). +""" +import re +import logging +import uuid +from typing import List, Dict, Any +from app.models.chunks import StructuredChunk +from app.utils.exceptions import ChunkingException + +logger = logging.getLogger(__name__) + + +class ChunkingService: + """Service for creating paragraph-based chunks from parsed document elements.""" + + def __init__(self, min_chunk_size: int = 50): + """ + Initialize chunking service. + + Args: + min_chunk_size: Minimum characters for a chunk (default 50) + """ + self.min_chunk_size = min_chunk_size + logger.info(f"Initialized ChunkingService: min_chunk_size={min_chunk_size}") + + def create_chunks( + self, + document_id: str, + parsed_elements: List[Dict[str, Any]] + ) -> List[StructuredChunk]: + """ + Create chunks by splitting on paragraph breaks (\\n\\n). + Each paragraph becomes a separate chunk for better RAG retrieval. + + Args: + document_id: Document UUID + parsed_elements: List of elements from PDF parsing + + Returns: + List of StructuredChunk objects + + Raises: + ChunkingException: If chunking fails + """ + try: + logger.info(f"Starting paragraph-based chunking: documentId={document_id}, elements={len(parsed_elements)}") + + # Step 1: Combine all elements into full text + full_text = "" + for element in parsed_elements: + text = element.get("text", "").strip() + if text: + full_text += text + "\n\n" + + if not full_text.strip(): + logger.warning("No text content found in parsed elements") + return [] + + # Step 2: Split by paragraph (\\n\\n) + paragraphs = full_text.split("\n\n") + + # Step 3: Create chunks from paragraphs + chunks = [] + chunk_index = 0 + + for paragraph in paragraphs: + paragraph = paragraph.strip() + + # Skip empty or very short paragraphs + if len(paragraph) < self.min_chunk_size: + logger.debug(f"Skipping short paragraph: length={len(paragraph)}") + continue + + # Create chunk + chunk_id = f"{document_id}-chunk-{chunk_index}" + + # Extract title (first line or first 50 chars) + title = self._extract_title(paragraph) + + chunk = StructuredChunk( + chunkId=chunk_id, + documentId=document_id, + content=paragraph, + title=title, + hierarchyLevel=1, + elementType="ParagraphChunk", + embedding=None, + metadata={ + "chunkIndex": chunk_index, + "length": len(paragraph), + "tokens": self._estimate_token_count(paragraph) + } + ) + + chunks.append(chunk) + chunk_index += 1 + + logger.debug( + f"Created chunk {chunk_index}: title='{title[:50]}...', " + f"length={len(paragraph)}, tokens=~{chunk.metadata['tokens']}" + ) + + # Log summary statistics + if chunks: + avg_length = sum(len(c.content) for c in chunks) / len(chunks) + avg_tokens = sum( + self._estimate_token_count(c.content) for c in chunks + ) / len(chunks) + + logger.info( + f"Chunking completed: documentId={document_id}, " + f"elements={len(parsed_elements)}, chunks={len(chunks)}, " + f"avgLength={int(avg_length)}, avgTokens={int(avg_tokens)}" + ) + + # Log each chunk summary + logger.info("Chunk summary:") + for i, chunk in enumerate(chunks): + tokens = self._estimate_token_count(chunk.content) + logger.info( + f" {i+1}. Title: '{chunk.title}' | " + f"Length: {len(chunk.content)} | Tokens: ~{tokens}" + ) + + return chunks + + except Exception as e: + logger.error(f"Chunking failed: documentId={document_id}", exc_info=True) + raise ChunkingException(f"Chunking failed: {str(e)}") + + def _extract_title(self, paragraph: str) -> str: + """ + Extract title from paragraph (first line or first 50 chars). + + Args: + paragraph: Paragraph text + + Returns: + Title string + """ + # Try to use first line as title + lines = paragraph.split('\n') + first_line = lines[0].strip() + + if len(first_line) > 0 and len(first_line) <= 100: + return first_line + + # Otherwise use first 50 characters + return paragraph[:50].strip() + ("..." if len(paragraph) > 50 else "") + + def _estimate_token_count(self, text: str) -> int: + """ + Estimate token count using simple heuristics. + Korean: ~1.5 tokens/char, English: ~1.3 tokens/word, Other: ~0.8 tokens/char + + Args: + text: Text to estimate tokens for + + Returns: + Estimated token count + """ + if not text: + return 0 + + # Count character types + korean_chars = sum(1 for c in text if '\uAC00' <= c <= '\uD7A3') + english_chars = sum(1 for c in text if ('a' <= c <= 'z') or ('A' <= c <= 'Z')) + other_chars = len(text) - korean_chars - english_chars + + # Count words for English + words = text.split() + word_count = len(words) + + # Estimate tokens + estimated_tokens = ( + word_count * 1.3 + + korean_chars * 1.5 + + other_chars * 0.8 + ) + + return max(1, int(round(estimated_tokens))) diff --git a/rag/app/services/embedding.py b/rag/app/services/embedding.py new file mode 100644 index 0000000..3dbbd5f --- /dev/null +++ b/rag/app/services/embedding.py @@ -0,0 +1,176 @@ +""" +Embedding service for generating semantic vectors using Ollama. +Uses LangChain Ollama integration for embeddings. +""" +import logging +import asyncio +from typing import List +from langchain_ollama import OllamaEmbeddings + +from app.models.chunks import StructuredChunk +from app.utils.exceptions import EmbeddingException + +logger = logging.getLogger(__name__) + + +class EmbeddingService: + """Service for generating embeddings using Ollama.""" + + def __init__( + self, + ollama_url: str, + model_name: str, + batch_size: int = 10 + ): + """ + Initialize embedding service. + + Args: + ollama_url: Ollama API URL + model_name: Embedding model name (e.g., dengcao/Qwen3-Embedding-0.6B:F16) + batch_size: Batch size for processing chunks + """ + self.ollama_url = ollama_url + self.model_name = model_name + self.batch_size = batch_size + + # Initialize Ollama embeddings + self.embeddings = OllamaEmbeddings( + base_url=ollama_url, + model=model_name + ) + + logger.info( + f"Initialized EmbeddingService: url={ollama_url}, " + f"model={model_name}, batch_size={batch_size}" + ) + + async def generate_embeddings( + self, + chunks: List[StructuredChunk] + ) -> List[StructuredChunk]: + """ + Generate embeddings for chunks in batches. + + Args: + chunks: List of StructuredChunk objects + + Returns: + List of chunks with embeddings attached + + Raises: + EmbeddingException: If embedding generation fails + """ + try: + logger.info(f"Starting embedding generation for {len(chunks)} chunks") + + embedded_chunks = [] + + # Process in batches + for i in range(0, len(chunks), self.batch_size): + batch = chunks[i:i + self.batch_size] + logger.debug( + f"Processing batch {i // self.batch_size + 1}: " + f"chunks {i}-{min(i + self.batch_size, len(chunks))}" + ) + + # Prepare texts for embedding (title + content) + texts = [self._prepare_text(chunk) for chunk in batch] + + # Generate embeddings (run synchronous call in thread pool) + try: + vectors = await asyncio.to_thread( + self.embeddings.embed_documents, + texts + ) + logger.debug( + f"Generated {len(vectors)} embeddings, " + f"dimension={len(vectors[0]) if vectors else 0}" + ) + except Exception as e: + logger.error(f"Ollama embedding generation failed: {str(e)}") + raise EmbeddingException(f"Ollama API error: {str(e)}") + + # Attach embeddings to chunks + for chunk, vector in zip(batch, vectors): + # Convert to List[float] (LangChain returns list already) + chunk.embedding = vector + embedded_chunks.append(chunk) + + logger.debug( + f"Embedded chunk: chunkId={chunk.chunkId}, " + f"vectorDim={len(vector)}" + ) + + logger.info( + f"Embedding generation completed: {len(embedded_chunks)} chunks embedded" + ) + + return embedded_chunks + + except EmbeddingException: + raise + except Exception as e: + logger.error("Embedding generation failed", exc_info=True) + raise EmbeddingException(f"Embedding generation failed: {str(e)}") + + async def generate_query_embedding(self, query: str) -> List[float]: + """ + Generate embedding for a search query. + + Args: + query: Search query text + + Returns: + Embedding vector + + Raises: + EmbeddingException: If embedding generation fails + """ + try: + logger.debug(f"Generating query embedding: query='{query[:50]}...'") + + # Generate query embedding (run synchronous call in thread pool) + vector = await asyncio.to_thread( + self.embeddings.embed_query, + query + ) + + logger.debug(f"Generated query embedding: dimension={len(vector)}") + + return vector + + except Exception as e: + logger.error("Query embedding generation failed", exc_info=True) + raise EmbeddingException(f"Query embedding failed: {str(e)}") + + def _prepare_text(self, chunk: StructuredChunk) -> str: + """ + Prepare text for embedding (title + content). + Truncates to max 8000 tokens (safe limit for most embedding models). + + Args: + chunk: Chunk to prepare + + Returns: + Combined text for embedding (truncated if needed) + """ + title = chunk.title if chunk.title else "" + content = chunk.content if chunk.content else "" + + if title: + text = f"Title: {title}\n{content}" + else: + text = content + + # Truncate to approximately 8000 tokens (assuming ~1.5 chars per token) + # This prevents Ollama from crashing on very long texts + MAX_CHARS = 12000 # ~8000 tokens + if len(text) > MAX_CHARS: + logger.warning( + f"Text too long ({len(text)} chars), truncating to {MAX_CHARS} chars " + f"for embedding (chunkId={chunk.chunkId})" + ) + text = text[:MAX_CHARS] + + return text diff --git a/rag/app/services/indexing.py b/rag/app/services/indexing.py new file mode 100644 index 0000000..485e4db --- /dev/null +++ b/rag/app/services/indexing.py @@ -0,0 +1,191 @@ +""" +Elasticsearch indexing service for storing document chunks. +Stores chunks in Elasticsearch ONLY (no PostgreSQL storage). +""" +import logging +from datetime import datetime +from typing import List +from elasticsearch import AsyncElasticsearch +from elasticsearch.helpers import async_bulk + +from app.models.chunks import StructuredChunk +from app.utils.exceptions import IndexingException + +logger = logging.getLogger(__name__) + + +class IndexingService: + """Service for indexing chunks to Elasticsearch.""" + + def __init__( + self, + es_client: AsyncElasticsearch, + index_name: str + ): + """ + Initialize indexing service. + + Args: + es_client: Elasticsearch async client + index_name: Index name for document chunks + """ + self.es = es_client + self.index_name = index_name + + logger.info(f"Initialized IndexingService: index={index_name}") + + async def index_chunks( + self, + chunks: List[StructuredChunk], + file_type: str = "UNKNOWN" + ) -> int: + """ + Bulk index chunks to Elasticsearch. + + Args: + chunks: List of StructuredChunk objects with embeddings + file_type: Document file type (PDF, MARKDOWN, TEXT) + + Returns: + Number of successfully indexed chunks + + Raises: + IndexingException: If indexing fails + """ + try: + logger.info(f"Starting bulk indexing: {len(chunks)} chunks") + + if not chunks: + logger.warning("No chunks to index") + return 0 + + # Prepare bulk actions + actions = [] + for chunk in chunks: + action = { + "_index": self.index_name, + "_id": chunk.chunkId, + "_source": self._prepare_document(chunk, file_type) + } + actions.append(action) + + # Perform bulk indexing + success_count, failed_items = await async_bulk( + self.es, + actions, + raise_on_error=False + ) + + if failed_items: + logger.error(f"Bulk indexing partial failure: {len(failed_items)} failed") + # Log first few failures for debugging + for item in failed_items[:5]: + logger.error(f"Failed item: {item}") + + logger.info( + f"Bulk indexing completed: success={success_count}, " + f"failed={len(failed_items)}" + ) + + # Refresh index to make documents immediately searchable + await self.es.indices.refresh(index=self.index_name) + + return success_count + + except Exception as e: + logger.error("Elasticsearch indexing failed", exc_info=True) + raise IndexingException(f"Indexing failed: {str(e)}") + + def _prepare_document( + self, + chunk: StructuredChunk, + file_type: str + ) -> dict: + """ + Prepare Elasticsearch document from chunk. + + Args: + chunk: StructuredChunk object + file_type: Document file type + + Returns: + Elasticsearch document dictionary + """ + return { + "chunkId": chunk.chunkId, + "sourceDocumentId": chunk.documentId, + "content": chunk.content, + "embedding": chunk.embedding, # 1024-dimensional vector + "indexedAt": datetime.utcnow().isoformat(), + "metadata": { + "title": chunk.title, + "hierarchyLevel": chunk.hierarchyLevel, + "fileType": file_type, + "language": "ko", # Korean as primary language + "elementType": chunk.elementType, + **chunk.metadata # Include original metadata + } + } + + async def delete_by_document_id(self, document_id: str) -> int: + """ + Delete all chunks for a document. + + Args: + document_id: Document UUID + + Returns: + Number of deleted chunks + + Raises: + IndexingException: If deletion fails + """ + try: + logger.info(f"Deleting chunks for document: {document_id}") + + query = { + "query": { + "term": { + "sourceDocumentId": document_id + } + } + } + + response = await self.es.delete_by_query( + index=self.index_name, + body=query + ) + + deleted_count = response.get("deleted", 0) + logger.info(f"Deleted {deleted_count} chunks for document {document_id}") + + return deleted_count + + except Exception as e: + logger.error(f"Deletion failed for document {document_id}", exc_info=True) + raise IndexingException(f"Deletion failed: {str(e)}") + + async def get_chunk_by_id(self, chunk_id: str) -> dict: + """ + Retrieve chunk by ID. + + Args: + chunk_id: Chunk identifier + + Returns: + Chunk document + + Raises: + IndexingException: If retrieval fails + """ + try: + response = await self.es.get( + index=self.index_name, + id=chunk_id + ) + + return response["_source"] + + except Exception as e: + logger.error(f"Failed to retrieve chunk: {chunk_id}", exc_info=True) + raise IndexingException(f"Chunk retrieval failed: {str(e)}") diff --git a/rag/app/services/parsing.py b/rag/app/services/parsing.py new file mode 100644 index 0000000..396211d --- /dev/null +++ b/rag/app/services/parsing.py @@ -0,0 +1,192 @@ +""" +Document parsing service using LangChain PyMuPDF for PDF processing. +Downloads files from presigned URLs and parses them into structured elements. +""" +import logging +import httpx +import tempfile +import os +from typing import List, Dict, Any +from pathlib import Path + +from langchain_community.document_loaders import PyMuPDFLoader +from app.utils.exceptions import DocumentParsingException, FileDownloadException + +logger = logging.getLogger(__name__) + + +class DocumentParsingService: + """Service for parsing PDF documents using PyMuPDF.""" + + def __init__(self): + """Initialize parsing service.""" + logger.info("Initialized DocumentParsingService with PyMuPDF") + + async def parse_document( + self, + file_url: str, + filename: str, + file_type: str + ) -> List[Dict[str, Any]]: + """ + Parse PDF document using PyMuPDF. + + Args: + file_url: Presigned MinIO URL for downloading the file + filename: Original filename + file_type: Document type (currently only PDF is supported) + + Returns: + List of parsed elements with type and text (compatible with chunking service) + + Raises: + FileDownloadException: If file download fails + DocumentParsingException: If parsing fails + """ + temp_file_path = None + + try: + logger.info(f"Starting PDF parsing: filename={filename}, type={file_type}") + + if file_type != "PDF": + raise DocumentParsingException( + f"Only PDF files are currently supported. Got: {file_type}" + ) + + # Download file from presigned URL + file_bytes = await self._download_file(file_url) + logger.info(f"Downloaded file: size={len(file_bytes)} bytes") + + # Save to temporary file (PyMuPDFLoader requires file path) + temp_file_path = await self._save_to_temp_file(file_bytes, filename) + + # Parse PDF using PyMuPDF + elements = await self._parse_pdf(temp_file_path, filename) + + logger.info(f"Parsing completed: filename={filename}, elements={len(elements)}") + + return elements + + except FileDownloadException: + raise + except DocumentParsingException: + raise + except Exception as e: + logger.error(f"Unexpected parsing error: filename={filename}", exc_info=True) + raise DocumentParsingException(f"Parsing failed: {str(e)}") + + finally: + # Clean up temporary file + if temp_file_path and os.path.exists(temp_file_path): + try: + os.unlink(temp_file_path) + logger.debug(f"Cleaned up temp file: {temp_file_path}") + except Exception as e: + logger.warning(f"Failed to clean up temp file: {e}") + + async def _download_file(self, file_url: str) -> bytes: + """ + Download file from presigned URL. + + Args: + file_url: Presigned MinIO URL + + Returns: + File bytes + + Raises: + FileDownloadException: If download fails + """ + try: + async with httpx.AsyncClient(timeout=300.0) as client: + response = await client.get(file_url) + response.raise_for_status() + return response.content + + except httpx.HTTPStatusError as e: + logger.error(f"File download failed: status={e.response.status_code}, url={file_url}") + raise FileDownloadException( + f"File download failed with status {e.response.status_code}" + ) + except Exception as e: + logger.error(f"File download error: {str(e)}") + raise FileDownloadException(f"File download failed: {str(e)}") + + async def _save_to_temp_file(self, file_bytes: bytes, filename: str) -> str: + """ + Save file bytes to a temporary file. + + Args: + file_bytes: File content + filename: Original filename (used for extension) + + Returns: + Path to temporary file + + Raises: + DocumentParsingException: If file save fails + """ + try: + # Get file extension + suffix = Path(filename).suffix or ".pdf" + + # Create temporary file + with tempfile.NamedTemporaryFile( + mode='wb', + suffix=suffix, + delete=False + ) as temp_file: + temp_file.write(file_bytes) + temp_path = temp_file.name + + logger.debug(f"Saved to temp file: {temp_path}") + return temp_path + + except Exception as e: + logger.error(f"Failed to save temp file: {str(e)}") + raise DocumentParsingException(f"Failed to save temp file: {str(e)}") + + async def _parse_pdf(self, file_path: str, filename: str) -> List[Dict[str, Any]]: + """ + Parse PDF using PyMuPDF and convert to element format. + + Args: + file_path: Path to PDF file + filename: Original filename + + Returns: + List of parsed elements compatible with chunking service + + Raises: + DocumentParsingException: If parsing fails + """ + try: + # Load PDF with PyMuPDF + loader = PyMuPDFLoader(file_path) + documents = loader.load() + + logger.debug(f"PyMuPDF loaded {len(documents)} pages") + + # Convert LangChain documents to elements format + elements = [] + for i, doc in enumerate(documents): + # Each page becomes a "NarrativeText" element + element = { + "type": "NarrativeText", + "text": doc.page_content, + "metadata": { + "page_number": i + 1, + "filename": filename, + "source": file_path, + **doc.metadata + } + } + elements.append(element) + + logger.info(f"Converted {len(elements)} pages to elements") + + return elements + + except Exception as e: + logger.error(f"PDF parsing failed: {str(e)}", exc_info=True) + raise DocumentParsingException(f"PDF parsing failed: {str(e)}") diff --git a/rag/app/services/search.py b/rag/app/services/search.py new file mode 100644 index 0000000..968c979 --- /dev/null +++ b/rag/app/services/search.py @@ -0,0 +1,248 @@ +""" +Hybrid search service combining BM25 keyword search and vector similarity. +Replicates Java SearchService logic. +""" +import logging +from typing import List +from elasticsearch import AsyncElasticsearch + +from app.services.embedding import EmbeddingService +from app.models.responses import SearchResultItem +from app.utils.exceptions import SearchException, ChunkNotFoundException + +logger = logging.getLogger(__name__) + + +class SearchService: + """Service for hybrid search (BM25 + Vector similarity).""" + + def __init__( + self, + es_client: AsyncElasticsearch, + index_name: str, + embedding_service: EmbeddingService, + bm25_weight: float = 0.3, + vector_weight: float = 0.7, + snippet_max_length: int = 50 + ): + """ + Initialize search service. + + Args: + es_client: Elasticsearch async client + index_name: Index name for searching + embedding_service: Service for generating query embeddings + bm25_weight: Weight for BM25 keyword search (default: 0.3) + vector_weight: Weight for vector similarity (default: 0.7) + snippet_max_length: Maximum length for snippets (default: 50) + """ + self.es = es_client + self.index_name = index_name + self.embedding_service = embedding_service + self.bm25_weight = bm25_weight + self.vector_weight = vector_weight + self.snippet_max_length = snippet_max_length + + logger.info( + f"Initialized SearchService: index={index_name}, " + f"bm25Weight={bm25_weight}, vectorWeight={vector_weight}" + ) + + async def search( + self, + query: str, + top_k: int = 50 + ) -> List[SearchResultItem]: + """ + Perform hybrid search (BM25 + Vector). + + Args: + query: Search query text + top_k: Number of top results to return + + Returns: + List of search result items + + Raises: + SearchException: If search fails + """ + try: + logger.info(f"Starting hybrid search: query='{query[:50]}...', topK={top_k}") + + # Generate query embedding + query_vector = await self.embedding_service.generate_query_embedding(query) + logger.debug(f"Generated query embedding: dimension={len(query_vector)}") + + # Build hybrid query + search_query = { + "size": top_k, + "query": { + "bool": { + "should": [ + # BM25 keyword search + { + "multi_match": { + "query": query, + "fields": ["content^2", "metadata.title^1.5"], + "type": "best_fields", + "fuzziness": "AUTO", + "boost": self.bm25_weight + } + }, + # Vector similarity search + { + "script_score": { + "query": {"match_all": {}}, + "script": { + "source": "(cosineSimilarity(params.query_vector, 'embedding') + 1.0) * params.vector_weight", + "params": { + "query_vector": query_vector, + "vector_weight": self.vector_weight + } + } + } + } + ] + } + }, + "_source": ["chunkId", "metadata.title", "content", "sourceDocumentId"], + "sort": [{"_score": {"order": "desc"}}] + } + + # Execute search + response = await self.es.search( + index=self.index_name, + body=search_query + ) + + hits = response["hits"]["hits"] + logger.debug(f"Elasticsearch returned {len(hits)} results") + + # Parse results + results = self._parse_search_results(hits) + + logger.info(f"Search completed: query='{query[:50]}...', results={len(results)}") + + return results + + except Exception as e: + logger.error(f"Search failed: query='{query[:50]}...'", exc_info=True) + raise SearchException(f"Search failed: {str(e)}") + + async def get_content( + self, + chunk_id: str, + max_tokens: int = 25000 + ) -> tuple[str, int]: + """ + Retrieve full content of a chunk with token limiting. + + Args: + chunk_id: Chunk identifier + max_tokens: Maximum tokens to return + + Returns: + Tuple of (content, actual_token_count) + + Raises: + ChunkNotFoundException: If chunk not found + SearchException: If retrieval fails + """ + try: + logger.info(f"Retrieving content: chunkId={chunk_id}, maxTokens={max_tokens}") + + # Get chunk from Elasticsearch + try: + response = await self.es.get( + index=self.index_name, + id=chunk_id + ) + source = response["_source"] + except Exception as e: + logger.error(f"Chunk not found: {chunk_id}") + raise ChunkNotFoundException(chunk_id) + + content = source.get("content", "") + + # Import here to avoid circular dependency + from app.utils.token_counter import token_counter + + # Count tokens + token_count = token_counter.count_tokens(content) + + # Truncate if necessary + if token_count > max_tokens: + logger.info( + f"Truncating content: original={token_count} tokens, " + f"limit={max_tokens} tokens" + ) + content, token_count = token_counter.truncate_to_token_limit( + content, + max_tokens, + preserve_beginning=True + ) + + logger.info( + f"Content retrieved: chunkId={chunk_id}, " + f"length={len(content)}, tokens={token_count}" + ) + + return content, token_count + + except ChunkNotFoundException: + raise + except Exception as e: + logger.error(f"Content retrieval failed: chunkId={chunk_id}", exc_info=True) + raise SearchException(f"Content retrieval failed: {str(e)}") + + def _parse_search_results(self, hits: List[dict]) -> List[SearchResultItem]: + """ + Parse Elasticsearch hits into SearchResultItem objects. + + Args: + hits: Elasticsearch search hits + + Returns: + List of SearchResultItem objects + """ + if not hits: + return [] + + # Find max score for normalization + max_score = max(hit["_score"] for hit in hits) if hits else 1.0 + + results = [] + for hit in hits: + source = hit["_source"] + score = hit["_score"] + + # Normalize score to 0.0-1.0 range + normalized_score = score / max_score if max_score > 0 else 0.0 + + result = SearchResultItem( + chunkId=source.get("chunkId", ""), + title=source.get("metadata", {}).get("title", "제목 없음"), + snippet=self._generate_snippet(source.get("content", "")), + relevanceScore=round(normalized_score, 4) + ) + results.append(result) + + return results + + def _generate_snippet(self, content: str) -> str: + """ + Generate snippet from content (first N characters). + + Args: + content: Full content + + Returns: + Snippet string + """ + if not content: + return "" + + if len(content) <= self.snippet_max_length: + return content + + return content[:self.snippet_max_length] + "..." diff --git a/rag/app/utils/__init__.py b/rag/app/utils/__init__.py new file mode 100644 index 0000000..feddb93 --- /dev/null +++ b/rag/app/utils/__init__.py @@ -0,0 +1 @@ +# Utils module diff --git a/rag/app/utils/__pycache__/__init__.cpython-312.pyc b/rag/app/utils/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..bb0c62e Binary files /dev/null and b/rag/app/utils/__pycache__/__init__.cpython-312.pyc differ diff --git a/rag/app/utils/__pycache__/exceptions.cpython-312.pyc b/rag/app/utils/__pycache__/exceptions.cpython-312.pyc new file mode 100644 index 0000000..dadaa9c Binary files /dev/null and b/rag/app/utils/__pycache__/exceptions.cpython-312.pyc differ diff --git a/rag/app/utils/exceptions.py b/rag/app/utils/exceptions.py new file mode 100644 index 0000000..90844f7 --- /dev/null +++ b/rag/app/utils/exceptions.py @@ -0,0 +1,68 @@ +""" +Custom exceptions for OpenContext RAG Service. +""" +from fastapi import HTTPException, status + + +class RagServiceException(Exception): + """Base exception for RAG service errors.""" + + def __init__(self, message: str, status_code: int = status.HTTP_500_INTERNAL_SERVER_ERROR): + self.message = message + self.status_code = status_code + super().__init__(self.message) + + +class DocumentParsingException(RagServiceException): + """Exception raised when document parsing fails.""" + + def __init__(self, message: str = "Document parsing failed"): + super().__init__(message, status.HTTP_500_INTERNAL_SERVER_ERROR) + + +class ChunkingException(RagServiceException): + """Exception raised when chunking fails.""" + + def __init__(self, message: str = "Document chunking failed"): + super().__init__(message, status.HTTP_500_INTERNAL_SERVER_ERROR) + + +class EmbeddingException(RagServiceException): + """Exception raised when embedding generation fails.""" + + def __init__(self, message: str = "Embedding generation failed"): + super().__init__(message, status.HTTP_500_INTERNAL_SERVER_ERROR) + + +class IndexingException(RagServiceException): + """Exception raised when Elasticsearch indexing fails.""" + + def __init__(self, message: str = "Elasticsearch indexing failed"): + super().__init__(message, status.HTTP_500_INTERNAL_SERVER_ERROR) + + +class SearchException(RagServiceException): + """Exception raised when search fails.""" + + def __init__(self, message: str = "Search operation failed"): + super().__init__(message, status.HTTP_500_INTERNAL_SERVER_ERROR) + + +class ChunkNotFoundException(RagServiceException): + """Exception raised when chunk is not found.""" + + def __init__(self, chunk_id: str): + message = f"Chunk not found: {chunk_id}" + super().__init__(message, status.HTTP_404_NOT_FOUND) + + +class FileDownloadException(RagServiceException): + """Exception raised when file download from presigned URL fails.""" + + def __init__(self, message: str = "Failed to download file from presigned URL"): + super().__init__(message, status.HTTP_500_INTERNAL_SERVER_ERROR) + + +def rag_exception_handler(exc: RagServiceException) -> HTTPException: + """Convert RAG service exceptions to HTTP exceptions.""" + return HTTPException(status_code=exc.status_code, detail=exc.message) diff --git a/rag/app/utils/token_counter.py b/rag/app/utils/token_counter.py new file mode 100644 index 0000000..d1f5889 --- /dev/null +++ b/rag/app/utils/token_counter.py @@ -0,0 +1,133 @@ +""" +Token counting utilities using tiktoken (cl100k_base encoder). +Provides accurate token counting for content retrieval with truncation support. +""" +import tiktoken +from typing import Optional + + +class TokenCounter: + """Token counter using tiktoken cl100k_base encoder.""" + + def __init__(self, encoding_name: str = "cl100k_base"): + """ + Initialize token counter with specified encoding. + + Args: + encoding_name: Tiktoken encoding name (default: cl100k_base for GPT-4) + """ + self.encoding = tiktoken.get_encoding(encoding_name) + self.encoding_name = encoding_name + + def count_tokens(self, text: str) -> int: + """ + Count tokens in text using tiktoken. + + Args: + text: Text to count tokens for + + Returns: + Number of tokens + """ + if not text: + return 0 + return len(self.encoding.encode(text)) + + def truncate_to_token_limit( + self, + text: str, + max_tokens: int, + preserve_beginning: bool = True + ) -> tuple[str, int]: + """ + Truncate text to fit within token limit using binary search. + Replicates Java ContentRetrievalService logic. + + Args: + text: Text to truncate + max_tokens: Maximum tokens allowed + preserve_beginning: If True, keep beginning of text; if False, keep end + + Returns: + Tuple of (truncated_text, actual_token_count) + """ + if not text: + return "", 0 + + # Check if truncation is needed + total_tokens = self.count_tokens(text) + if total_tokens <= max_tokens: + return text, total_tokens + + # Binary search to find maximum length that fits within token limit + left, right = 0, len(text) + best_length = 0 + + while left <= right: + mid = (left + right) // 2 + if preserve_beginning: + candidate = text[:mid] + else: + candidate = text[-mid:] + + token_count = self.count_tokens(candidate) + + if token_count <= max_tokens: + best_length = mid + left = mid + 1 + else: + right = mid - 1 + + # Return text at best length + if preserve_beginning: + truncated = text[:best_length] + else: + truncated = text[-best_length:] + + actual_tokens = self.count_tokens(truncated) + return truncated, actual_tokens + + def estimate_tokens(self, text: str) -> int: + """ + Estimate tokens using simplified heuristics (fallback method). + Replicates Java ChunkingService token estimation logic. + + Args: + text: Text to estimate tokens for + + Returns: + Estimated token count + """ + if not text: + return 0 + + # Count different character types + korean_chars = sum( + 1 for c in text + if '\uAC00' <= c <= '\uD7A3' # Hangul syllables + ) + english_chars = sum( + 1 for c in text + if ('a' <= c <= 'z') or ('A' <= c <= 'Z') + ) + other_chars = len(text) - korean_chars - english_chars + + # Word count for English text + words = text.split() + word_count = len(words) + + # Estimate tokens + # - English: ~1.3 tokens per word + # - Korean: ~1.5 tokens per character + # - Other: ~0.8 tokens per character + estimated_tokens = ( + word_count * 1.3 + + korean_chars * 1.5 + + other_chars * 0.8 + ) + + return max(1, int(round(estimated_tokens))) + + +# Global token counter instance +token_counter = TokenCounter() diff --git a/rag/requirements.txt b/rag/requirements.txt new file mode 100644 index 0000000..7d17af9 --- /dev/null +++ b/rag/requirements.txt @@ -0,0 +1,25 @@ +# FastAPI Framework +fastapi==0.115.0 +uvicorn[standard]==0.32.0 +pydantic==2.10.3 +pydantic-settings==2.6.1 + +# LangChain Python +langchain==0.3.11 +langchain-community==0.3.11 +langchain-ollama==0.2.0 + +# PDF Processing +pymupdf==1.24.14 +langchain-text-splitters==0.3.2 + +# Elasticsearch +elasticsearch>=9.1.0,<10.0.0 + +# HTTP Clients +httpx==0.28.1 + +# Utilities +python-dotenv==1.0.1 +tiktoken==0.8.0 +python-multipart==0.0.12