Skip to content

bast-ai-agno

Version: 0.3.2 Module: bast_ai_agno Path: src/python/agno

Overview

Comprehensive agent configuration and workflow management system for Agno agents. Provides pre-configured agent setups, knowledge retrieval with MongoDB, workflow orchestration, and integration utilities.

Features

Agent Configuration System

  • Pre-configured agent configurations for common use cases
  • Configuration utilities for building custom configurations
  • Validation functions for configuration validation
  • Merge utilities for combining configurations
  • Configuration presets for optimized agents

Workflow System

  • Pre-configured workflow configurations
  • Workflow factory for building workflows from templates
  • Separate MongoDB storage for workflow execution data
  • Support for agents, teams, and functions as workflow steps

BastRetrieverMongo

Intelligent knowledge retrieval with hybrid search capabilities.

from bast_ai_agno.svc import BastRetrieverMongo

retriever = BastRetrieverMongo(
    catname="advisor",
    prefer="hybrid",              # "text", "vector", or "hybrid"
    use_atlas_search=True,
    use_vector_search=True
)

results = retriever.knowledge_retriever(
    "search query",
    num_documents=5,
    entities=["entity1", "entity2"],
    accepted_score=80.0
)

Search Modes

Mode Description Best For
text MongoDB text indexes, Atlas Search Exact keyword matches
vector Embedding similarity, Atlas Vector Search Semantic similarity
hybrid Combines both methods Comprehensive coverage

BastJSONReader

Enhanced JSON reader for knowledge base operations.

from bast_ai_agno.svc.bast_json_reader import BastJSONReader
from agno.knowledge.chunking.fixed import FixedSizeChunking

reader = BastJSONReader(chunking_strategy=FixedSizeChunking(chunk_size=1000))
documents = reader.read("path/to/data.json")

# Async support
documents = await reader.async_read("path/to/data.json")

Supported JSON Formats

  • Direct arrays: [{...}, {...}]
  • Wrapped arrays: {"items": [...]}, {"pages": [...]}, {"records": [...]}
  • Single objects: {...}

Chunking Strategies

  • FixedSizeChunking: Fixed-size chunks
  • AgenticChunking: AI-powered semantic chunks
  • DocumentChunking: Document boundary splits
  • RecursiveChunking: Recursive splitting
  • SemanticChunking: Semantic similarity splits

Critical Agent Rules

Never Create Agents in Loops

# WRONG - massive performance overhead
for query in queries:
    agent = Agent(...)  # DON'T DO THIS

# CORRECT
agent = Agent(...)
for query in queries:
    agent.run(query)

Always Use output_schema

from pydantic import BaseModel

class Response(BaseModel):
    answer: str
    confidence: float

agent = Agent(
    model=OpenAIChat(id="gpt-4o"),
    output_schema=Response
)

Agent Pattern Selection

Pattern Use When Example
Single Agent (90%) One clear task/domain FAQ bot, data analyzer
Team Multiple specialists needed Research + writing
Workflow Sequential steps with logic ETL pipeline

Dependencies

Internal Dependencies

Package Purpose
bast-ai-config Configuration management

External Dependencies

Package Version Purpose
agno * Agent framework (with tavily extras)
pypdf 6.0.0 PDF parsing

Testing

Unit Tests (11 files)

  • test_svc_api.py - Service API tests
  • test_agent_teams_mongo_api.py - Team MongoDB tests
  • test_bast_json_reader.py - JSON reader tests
  • test_agno_teams.py - Team functionality tests
  • test_tool_*.py - Tool-specific tests
  • test_bast_retriever_mongo.py - Retriever tests
  • test_web_search_*.py - Web search tests

Integration Tests (1 file)

  • test_circular_imports.py - Import validation

Running Tests

pants test --output=all src/python/agno:: -- -s --no-header

Documentation

Comprehensive documentation available in src/python/agno/docs/:

Issues & Improvements

Current Issues

  1. Limited async support in retriever: Some operations are synchronous
  2. MongoDB Atlas dependency: Full features require Atlas Search/Vector Search

Suggested Improvements

  1. Caching layer: Add Redis caching for embeddings
  2. Batch processing: Support batch queries in retriever
  3. More chunking strategies: Add custom chunking support
  4. Metrics collection: Add performance metrics for retrieval

Usage Examples

Basic Agent Setup

from agno.agent import Agent
from agno.models.openai import OpenAIChat

agent = Agent(
    model=OpenAIChat(id="gpt-4o"),
    instructions="You are a helpful assistant",
    markdown=True,
)
agent.print_response("Your query", stream=True)

Knowledge Retrieval

from bast_ai_agno.svc import BastRetrieverMongo

retriever = BastRetrieverMongo(
    catname="advisor",
    prefer="hybrid",
    use_atlas_search=True,
    use_vector_search=True,
    mongo_max_time_ms=500,
    sample_size=50
)

results = retriever.knowledge_retriever(
    "international student advising",
    num_documents=5,
    entities=["MSU Denver", "F-1 visa"],
    accepted_score=80.0
)

for doc in results:
    print(f"Score: {doc['score_norm']:.1f}")
    print(f"Name: {doc['name']}")
    print(f"Content: {doc['content'][:100]}...")

Using AgnoKnowledgeResources

from bast_ai_agno.bp.kb_resources import AgnoKnowledgeResources

kb_resources = AgnoKnowledgeResources("your_cat_name")
json_reader = kb_resources.get_bast_json_reader()

knowledge = kb_resources.get_knowledge(
    kb_name="my_kb",
    reader=json_reader
)

Changelog

v0.3.2 (Current)

  • Enhanced BastJSONReader with async support
  • Improved score normalization in retriever
  • Added workflow factory

v0.3.1

  • Added BastRetrieverMongo hybrid search
  • Configuration validation improvements

v0.3.0

  • Major refactor of agent configuration system
  • Added workflow support