diff --git a/README.md b/README.md index b67ce4081..d2686fbfa 100644 --- a/README.md +++ b/README.md @@ -89,7 +89,7 @@ A: Yes. Knowhere extracts them, runs them through VLMs for summarization and fea ## Performance Benchmark -Agents using Knowhere outperform those working from raw documents or MinerU-parsed output on real-world tasks: searching, modifying, and answering questions. +Agents using Knowhere outperform those working from raw documents, MinerU-parsed output, or Unstructured output on real-world tasks: searching, modifying, and answering questions.

Benchmark Performance: Agent + Knowhere vs Others @@ -99,11 +99,11 @@ Agents using Knowhere outperform those working from raw documents or MinerU-pars ### Key Advantages -- **Accuracy**: +36% first-try accuracy and +10% recall over raw documents. +- **Accuracy**: +36% first-try accuracy and +11% recall over raw documents. - **Reliability**: 79% accuracy with feedback, vs. a ~53% ceiling on raw docs. - **Efficiency**: Fewer loops, fewer tokens, less time. Agents navigate a structured graph instead of reading monolithic text. -*(Internal evaluation across identical agentic RAG tasks. Baseline: MinerU output fed directly to agents.)* +*(Internal evaluation across identical agentic RAG tasks. Baselines: raw documents, MinerU output, and Unstructured output fed directly to agents.)* > [!NOTE] > **📊 Benchmarks are actively expanding.** More parsers and retrieval baselines coming soon. diff --git a/apps/worker/app/services/document_ingestion/processing_billing.py b/apps/worker/app/services/document_ingestion/processing_billing.py index 94a3d77ad..6fd6cc321 100644 --- a/apps/worker/app/services/document_ingestion/processing_billing.py +++ b/apps/worker/app/services/document_ingestion/processing_billing.py @@ -129,8 +129,9 @@ def record_processing_start( billing_snapshot: ParseJobBillingSnapshot, processing_started_at: datetime, workload_estimate: WorkloadEstimate, + extra_metadata: dict[str, object] | None = None, ) -> None: - metadata_updates = { + metadata_updates: dict[str, object] = { "page_count": workload_estimate.page_count, "billing_status": billing_snapshot.billing_status, "billing_amount_micro_dollars": billing_snapshot.billing_amount_micro_dollars, @@ -142,6 +143,8 @@ def record_processing_start( metadata_updates["workload_estimate_fallback_reason"] = ( workload_estimate.fallback_reason ) + if extra_metadata is not None: + metadata_updates.update(extra_metadata) with get_sync_db_context() as db: job_result = db.execute(select(Job).where(Job.job_id == job_id).with_for_update()) job = job_result.scalar_one_or_none() diff --git a/apps/worker/app/services/document_ingestion/processing_run.py b/apps/worker/app/services/document_ingestion/processing_run.py index 69072d51d..21e73befd 100644 --- a/apps/worker/app/services/document_ingestion/processing_run.py +++ b/apps/worker/app/services/document_ingestion/processing_run.py @@ -25,7 +25,7 @@ ) from loguru import logger -from shared.core.exceptions.domain_exceptions import ValidationException +from shared.core.exceptions.domain_exceptions import ValidationException, Violation from shared.services.jobs.lifecycle.service import get_sync_job_lifecycle_service from shared.services.redis.distributed_lock import RedisJobLock from shared.services.redis.redis_sync_service import ( @@ -97,6 +97,7 @@ def _run_parse_job( file_extension=prepared_source.file_extension, page_count=page_count, ): + violations = _build_pdf_page_limit_violations(page_count) billing_snapshot = record_skipped_parse_job_billing( job_id=job_id, workload_estimate=workload_estimate, @@ -107,8 +108,9 @@ def _run_parse_job( billing_snapshot=billing_snapshot, processing_started_at=processing_started_at, workload_estimate=workload_estimate, + extra_metadata={"error_details": {"violations": violations}}, ) - _raise_pdf_page_limit_exceeded(page_count) + _raise_pdf_page_limit_exceeded(page_count, violations) billing_snapshot = charge_parse_job_pages( job_id=job_id, @@ -168,7 +170,21 @@ def _is_pdf_page_limit_exceeded(*, file_extension: str, page_count: int) -> bool return file_extension == ".pdf" and page_count > settings.MAX_PDF_PAGE_LIMIT -def _raise_pdf_page_limit_exceeded(page_count: int) -> None: +def _build_pdf_page_limit_violations(page_count: int) -> list[Violation]: + from shared.core.config import settings + + pdf_page_limit = settings.MAX_PDF_PAGE_LIMIT + return [ + { + "field": "page_count", + "description": f"PDF has {page_count} pages, limit is {pdf_page_limit}", + } + ] + + +def _raise_pdf_page_limit_exceeded( + page_count: int, violations: list[Violation] +) -> None: from shared.core.config import settings pdf_page_limit = settings.MAX_PDF_PAGE_LIMIT @@ -177,10 +193,5 @@ def _raise_pdf_page_limit_exceeded(page_count: int) -> None: f"Document too large: {page_count} pages exceeds the {pdf_page_limit}-page limit. " "Please split the document and upload in smaller batches." ), - violations=[ - { - "field": "page_count", - "description": f"PDF has {page_count} pages, limit is {pdf_page_limit}", - } - ], + violations=violations, ) diff --git a/apps/worker/tests/contract/conftest.py b/apps/worker/tests/contract/conftest.py index c1550b145..41ba3846a 100644 --- a/apps/worker/tests/contract/conftest.py +++ b/apps/worker/tests/contract/conftest.py @@ -9,6 +9,7 @@ import pytest from pytest_postgresql import factories +from celery import Celery from pytest import MonkeyPatch from shared.testing import contract_runtime from shared.testing.contract_runtime import PostgreSQLProcess @@ -16,6 +17,12 @@ _REPO_ROOT: Path = Path(__file__).resolve().parents[4] _WORKER_ROOT: Path = _REPO_ROOT / "apps" / "worker" +_DOCUMENT_INGESTION_TASK_NAMES: tuple[str, ...] = ( + "app.core.tasks.document_ingestion_tasks.upload_url_file_task", + "app.core.tasks.kb_tasks.upload_url_file_task", + "app.core.tasks.document_ingestion_tasks.parse_task", + "app.core.tasks.kb_tasks.parse_task", +) def _resolve_postgresql_executable() -> str | None: @@ -28,6 +35,11 @@ def _resolve_postgresql_executable() -> str | None: return str(executable_path) if executable_path is not None else None +def _clear_document_ingestion_task_registrations(celery_app: Celery) -> None: + for task_name in _DOCUMENT_INGESTION_TASK_NAMES: + celery_app.tasks.pop(task_name, None) + + _contract_postgresql_proc = factories.postgresql_proc( executable=_resolve_postgresql_executable(), port=contract_runtime.CONTRACT_POSTGRESQL_PORT_RANGE, @@ -64,6 +76,7 @@ def worker_contract_environment( celery_app = get_celery_app() monkeypatch.setattr(celery_app.conf, "task_always_eager", True) monkeypatch.setattr(celery_app.conf, "task_eager_propagates", False) + _clear_document_ingestion_task_registrations(celery_app) importlib.import_module("app.core.tasks.document_ingestion_tasks") try: diff --git a/docs/assets/benchmark.png b/docs/assets/benchmark.png index ed5b555f7..464f0f217 100644 Binary files a/docs/assets/benchmark.png and b/docs/assets/benchmark.png differ diff --git a/packages/shared-python/shared/services/redis/redis_sync_service.py b/packages/shared-python/shared/services/redis/redis_sync_service.py index 812b838f6..ec4d996d7 100644 --- a/packages/shared-python/shared/services/redis/redis_sync_service.py +++ b/packages/shared-python/shared/services/redis/redis_sync_service.py @@ -261,7 +261,6 @@ def close(self): self._client = None logger.info("Sync Redis client closed") - class SyncRedisServiceFactory: """Factory for sync Redis service instances."""