Version: 1.0.0 Last Updated: 2026-05-14 Status: Active

bast-agno-pipeline¶

Version: 0.1.4 Module: bast_agno_pipeline Path: src/pipelines/agno_pipeline

Overview¶

Microservice pipeline for ingesting JSON and PDF data into Agno Knowledge bases with support for metadata handling, filtering, and concurrent processing.

Features¶

Data Ingestion¶

JSON file ingestion with automatic parsing
PDF file ingestion (new in v0.1.4)
In-memory data support
Batch processing with configurable concurrency

Metadata Management¶

Intelligent metadata merging from multiple sources
Automatic filter generation based on source
Entity extraction and preservation
Custom metadata support

Performance¶

Asynchronous processing with asyncio
Configurable concurrent uploads
Real-time progress tracking
Duplicate detection and handling

Main Services¶

Service	Description
`JSONIngestPipeline`	Main ingestion pipeline
`KBCleanup`	Knowledge base cleanup utilities
`KBInspect`	KB inspection and diagnostics

Usage Examples¶

Basic JSON Ingestion¶

from bast_agno_pipeline.svc import JSONIngestPipeline
from bast_ai_agno import AgnoKnowledgeResources

# Get knowledge resources
resources = AgnoKnowledgeResources(catname='your_catname')
embedder = resources.get_openai_embedder()
knowledge = resources.get_knowledge('your_kb_name', embedder=embedder)

# Create pipeline
pipeline = JSONIngestPipeline(
    knowledge=knowledge,
    default_source="my_source",
    default_author="my_author"
)

# Ingest JSON file
await pipeline.add_json_content(
    "path/to/data.json",
    upsert=True,
    max_concurrency=10
)

PDF Ingestion¶

# Ingest PDF file (new in v0.1.4)
await pipeline.add_pdf_content(
    "path/to/document.pdf",
    upsert=True,
    max_concurrency=5
)

In-Memory Data Ingestion¶

documents = [
    {
        "content": "Document text here",
        "name": "Document 1",
        "meta_data": {"category": "tutorial"}
    },
    {
        "content": "Another document",
        "name": "Document 2",
        "meta_data": {"category": "reference"}
    }
]

await pipeline.add_documents(
    documents=documents,
    upsert=True
)

Progress Tracking¶

def progress_callback(current: int, total: int, message: str):
    percentage = (current / total) * 100
    print(f"Progress: {percentage:.1f}% - {message}")

await pipeline.add_json_content(
    "path/to/data.json",
    progress_callback=progress_callback
)

Knowledge Base Cleanup¶

from bast_agno_pipeline.svc import KBCleanup

cleanup = KBCleanup(catname="mycat", kb_name="my_kb")

# Remove documents by filter
await cleanup.remove_by_filter({"source": "old_source"})

# Clear all documents
await cleanup.clear_all()

KB Inspection¶

from bast_agno_pipeline.svc import KBInspect

inspect = KBInspect(catname="mycat", kb_name="my_kb")

# Get KB statistics
stats = await inspect.get_stats()
print(f"Total documents: {stats['total_documents']}")
print(f"Total embeddings: {stats['total_embeddings']}")

# Sample documents
samples = await inspect.get_samples(limit=5)

Dependencies¶

Internal Dependencies¶

Package	Purpose
bast-ai-base	Core utilities
bast-ai-config	Configuration management

Testing¶

Unit Tests (2 files)¶

test_init_mongo_collections.py - MongoDB collection tests
test_process_agno_data.py - Data processing tests

Integration Tests (3 files)¶

agno_kb_cleanup.py - KB cleanup tests
agno_kb_inspect.py - KB inspection tests
agno_kb_pipeline.py - Full pipeline test

Running Tests¶

# Unit tests
pants test src/pipelines/agno_pipeline:: -- -s --no-header

# Integration tests
pants run src/pipelines/agno_pipeline/integration/agno_kb_pipeline.py

Configuration¶

# resources/config/{catname}.yaml
agno_pipeline:
  default_concurrency: 10
  batch_size: 100
  duplicate_handling: skip  # or "update", "error"

  metadata_defaults:
    source: "default_source"
    author: "system"

  progress:
    report_interval: 10  # Report every N documents

JSON Input Format¶

Expected Structure¶

[
  {
    "content": "The main text content",
    "name": "Document title or identifier",
    "meta_data": {
      "source": "source_name",
      "author": "author_name",
      "category": "category_name",
      "entities": ["entity1", "entity2"],
      "custom_field": "custom_value"
    }
  }
]

Supported Wrapper Formats¶

// Wrapped in "items"
{"items": [...]}

// Wrapped in "data"
{"data": [...]}

// Wrapped in "documents"
{"documents": [...]}

Issues & Improvements¶

Current Issues¶

Memory usage: Large files loaded entirely into memory
Limited error recovery: Failed documents not retried
No incremental updates: Full re-ingestion required for updates

Suggested Improvements¶

Streaming ingestion: Process large files in chunks
Retry queue: Queue failed documents for retry
Incremental sync: Support delta updates
Checkpointing: Resume interrupted ingestion
Validation: Pre-validate documents before ingestion
Metrics: Add ingestion metrics and monitoring

Changelog¶

v0.1.4 (Current)¶

Added PDF file ingestion support
Unified API with single file_path parameter
Enhanced progress tracking for PDF processing
Added time estimates based on file size

v0.1.3¶

Improved metadata merging
Better duplicate detection

v0.1.2¶

Added progress callback support
Configurable concurrency

v0.1.1¶

Initial release with JSON ingestion