naas-abi-core 1.21.5

Abi framework allowing you to build your AI system.

naas-abi-core

The core implementation library for ABI (Agentic Brain Infrastructure), providing the fundamental building blocks for building unified AI systems. This library implements the core concepts, services, and architecture patterns that enable ontology-driven AI applications.

Overview

The ABI Library is the core implementation of ABI's concepts, designed to build a unified AI system. This library provides the fundamental building blocks for connecting, processing, and utilizing data across different AI components.

naas-abi-core is the foundational library that powers the ABI framework. It provides:

• Engine: Central orchestration system that loads and coordinates modules, services, and ontologies
• Services: Hexagonal architecture-based services for storage, secrets, and AI capabilities
• Modules: Modular system for organizing agents, integrations, pipelines, and workflows
• Agents: LangGraph-based AI agents with tool binding and conversation management
• Applications: Ready-to-use interfaces (REST API, Terminal, MCP Protocol)

Installation

pip install naas-abi-core

Optional Dependencies

# For Qdrant vector store support
pip install naas-abi-core[qdrant]

# For AWS S3 object storage support
pip install naas-abi-core[aws]

# For SSH tunnel support
pip install naas-abi-core[ssh]

# For OpenRouter integration
pip install naas-abi-core[openrouter]

# Install all optional dependencies
pip install naas-abi-core[all]

Core Architecture

Engine

The Engine is the central orchestrator that:

1. Loads Configuration: Reads and validates YAML configuration files
2. Initializes Services: Sets up storage, vector, triple store, and secret services based on module dependencies
3. Loads Modules: Discovers and loads modules with their agents, integrations, pipelines, and workflows
4. Loads Ontologies: Loads RDF ontologies into the triple store for semantic reasoning
5. Initializes Components: Calls on_initialized() on all modules after everything is loaded

Example Usage:

from naas_abi_core.engine.Engine import Engine

# Initialize engine with default configuration (config.yaml)
engine = Engine()

# Load all modules
engine.load()

# Or load specific modules
engine.load(module_names=["naas_abi", "my_custom_module"])

# Access loaded modules
for module_name, module in engine.modules.items():
    print(f"Module: {module_name}")
    print(f"Agents: {[agent.__name__ for agent in module.agents]}")

# Access services
triple_store = engine.services.triple_store
vector_store = engine.services.vector_store
object_storage = engine.services.object_storage
secret_service = engine.services.secret

Modules

Modules are the primary organizational unit in ABI. Each module can contain:

• Agents: AI agents that can be used for conversations and task execution
• Integrations: Connections to third-party services and APIs
• Pipelines: Data transformation processes that convert raw data into semantic representations
• Workflows: Business logic that can be exposed as tools, API endpoints, or scheduled jobs
• Ontologies: RDF/Turtle files that define semantic knowledge structures

Module Structure:

from naas_abi_core.module.Module import BaseModule, ModuleConfiguration
from naas_abi_core.engine.EngineProxy import EngineProxy

class MyModule(BaseModule):
    class Configuration(ModuleConfiguration):
        # Module-specific configuration
        api_key: str
    
    dependencies = ModuleDependencies(
        modules=[],  # Other modules this module depends on
        services=[TripleStoreService, VectorStoreService]  # Required services
    )
    
    def on_load(self):
        # Called when module is loaded
        # Load ontologies, agents, etc.
        super().on_load()
    
    def on_initialized(self):
        # Called after all modules and services are initialized
        # Safe to access other modules and services here
        pass

Services

Services form the foundational layer of ABI, implementing the Hexagonal Architecture (Ports & Adapters) pattern to provide flexible and system-agnostic interfaces. This architectural approach allows ABI to seamlessly integrate with existing systems while maintaining clean separation of concerns.

Each service defines a primary port (interface) that specifies its capabilities, while multiple secondary adapters can implement this interface for different backend systems. This means you can:

• Easily swap implementations without changing business logic
• Add new integrations by implementing new adapters
• Test components in isolation using mock adapters

For example, the Secret Service could connect to various backend systems through different adapters:

• Hashicorp Vault
• AWS Secrets Manager
• Azure Key Vault
• Environment Variables
• Local File System
• Google Cloud Secret Manager
• Kubernetes Secrets

This modular approach ensures that ABI can be deployed in any environment while maintaining consistent interfaces and behavior across different infrastructure choices.

Triple Store Service

Manages RDF knowledge graphs for semantic reasoning and ontology storage.

Supported Backends:

• Oxigraph (default)
• SPARQL endpoints
• Custom adapters

Example:

from naas_abi_core.services.triple_store.TripleStoreService import TripleStoreService

triple_store = engine.services.triple_store

# Query the knowledge graph
results = triple_store.query("""
    SELECT ?subject ?predicate ?object
    WHERE {
        ?subject ?predicate ?object
    }
    LIMIT 10
""")

Vector Store Service

Manages vector embeddings for semantic search and similarity matching.

Supported Backends:

• Qdrant (optional, requires [qdrant] extra)
• Custom adapters

Example:

from naas_abi_core.services.vector_store.VectorStoreService import VectorStoreService

vector_store = engine.services.vector_store

# Store embeddings
vector_store.upsert(
    collection_name="intents",
    vectors=[embedding],
    ids=["intent_1"],
    payloads=[{"text": "user query"}]
)

# Search similar vectors
results = vector_store.search(
    collection_name="intents",
    query_vector=query_embedding,
    limit=5
)

Object Storage Service

Manages file storage for documents, reports, and generated content.

Supported Backends:

• AWS S3 (optional, requires [aws] extra)
• MinIO
• Local file system
• Custom adapters

Example:

from naas_abi_core.services.object_storage.ObjectStorageService import ObjectStorageService

object_storage = engine.services.object_storage

# Upload a file
object_storage.upload(
    bucket="my-bucket",
    key="documents/report.pdf",
    file_path="/path/to/report.pdf"
)

# Download a file
object_storage.download(
    bucket="my-bucket",
    key="documents/report.pdf",
    file_path="/path/to/downloaded.pdf"
)

Secret Service

Manages secrets and credentials securely across different storage systems.

Supported Backends:

• Environment variables
• Naas Secret Manager
• Hashicorp Vault
• AWS Secrets Manager
• Azure Key Vault
• Google Cloud Secret Manager
• Kubernetes Secrets
• Local file system
• Custom adapters

Example:

from naas_abi_core.services.secret.Secret import Secret

secret_service = engine.services.secret

# Get a secret
api_key = secret_service.get("OPENAI_API_KEY")

# List all secrets
all_secrets = secret_service.list()

Cache Service

Provides intelligent caching for API calls, tool results, and model responses to optimize performance and manage rate limits.

Example:

from naas_abi_core.services.cache.CacheService import CacheService

cache = CacheService()

# Cache a function result
@cache.cache(ttl=3600)
def expensive_api_call(param: str):
    # Expensive operation
    return result

# Force refresh
result = expensive_api_call(param, force_refresh=True)

Core Concepts

Integration

Integrations provide standardized connections to third-party services and data sources. They handle:

• Authentication and authorization
• API communication
• Data format standardization
• Error handling and retries

Example:

from naas_abi_core.integration.integration import Integration, IntegrationConfiguration

class MyAPIConfiguration(IntegrationConfiguration):
    api_key: str
    base_url: str

class MyAPIIntegration(Integration):
    def __init__(self, configuration: MyAPIConfiguration):
        super().__init__(configuration)
        # Initialize connection
    
    def fetch_data(self):
        # Implement API call
        pass

Pipeline

Pipelines are responsible for data ingestion and transformation into the ontological layer. They:

• Utilize integrations to fetch data
• Transform raw data into semantic representations
• Maintain data consistency and quality
• Map external data models to ABI's ontology

Example:

from naas_abi_core.pipeline.pipeline import Pipeline, PipelineConfiguration, PipelineParameters
from rdflib import Graph

class MyPipelineConfiguration(PipelineConfiguration):
    integration_config: dict

class MyPipelineParameters(PipelineParameters):
    source: str

class MyPipeline(Pipeline):
    def __init__(self, configuration: MyPipelineConfiguration):
        super().__init__(configuration)
    
    def run(self, parameters: MyPipelineParameters) -> Graph:
        # Fetch data from integration
        # Transform to RDF
        # Return Graph
        graph = Graph()
        # ... add triples to graph
        return graph
    
    def trigger(self, event, ontology_name, triple) -> Graph:
        # Event-driven pipeline execution
        return self.run(MyPipelineParameters(source="event"))

Workflow

Workflows leverage the ontological layer to implement business logic and provide data to consumers. They can be used by:

• Large Language Models (LLMs)
• Remote APIs and services
• Other automated processes

Example:

from naas_abi_core.workflow.workflow import Workflow, WorkflowConfiguration, WorkflowParameters
from pydantic import BaseModel

class MyWorkflowParameters(WorkflowParameters):
    input_data: str

class MyWorkflowConfiguration(WorkflowConfiguration):
    processing_option: str

class MyWorkflow(Workflow[MyWorkflowParameters]):
    def __init__(self, configuration: MyWorkflowConfiguration):
        super().__init__(configuration)
    
    def run(self, parameters: MyWorkflowParameters):
        # Implement business logic
        # Query knowledge graph
        # Process data
        # Return results
        return {"result": "processed"}

Agent

Agents are AI-powered assistants that can have conversations, use tools, and delegate to sub-agents.

Features:

• LangGraph-based conversation management
• Tool binding and execution
• Sub-agent delegation
• Intent-based routing
• Conversation persistence (PostgreSQL checkpointing)
• Event streaming for real-time updates

Example:

from naas_abi_core.services.agent.Agent import Agent
from langchain_openai import ChatOpenAI

class MyAgent(Agent):
    def __init__(self):
        super().__init__(
            name="MyAgent",
            description="An agent that helps with specific tasks",
            system_prompt="You are a helpful assistant...",
            chat_model=ChatOpenAI(model="gpt-4"),
            tools=[my_tool, my_workflow],
            agents=[sub_agent]  # Optional sub-agents
        )

Applications

REST API

FastAPI-based REST API that automatically exposes all agents, workflows, and pipelines.

Features:

• OpenAPI/Swagger documentation
• OAuth2 authentication
• CORS support
• Automatic endpoint generation from agents/workflows/pipelines

Usage:

from naas_abi_core.apps.api.api import api

# Run the API server
api()

Endpoints:

• GET / - API landing page
• GET /docs - Swagger UI documentation
• GET /redoc - ReDoc documentation
• POST /agents/{agent_name}/completion - Agent completion endpoint
• POST /workflows/{workflow_name}/run - Workflow execution endpoint
• POST /pipelines/{pipeline_name}/run - Pipeline execution endpoint

Terminal Agent

Interactive terminal interface for chatting with agents.

Usage:

from naas_abi_core.apps.terminal_agent.main import run_agent
from naas_abi_core.services.agent.Agent import Agent

agent = MyAgent()
run_agent(agent)

MCP Server

Model Context Protocol (MCP) server for integration with Claude Desktop and VS Code.

Features:

• Dynamic agent discovery from OpenAPI spec
• HTTP and stdio transport modes
• Automatic tool registration

Usage:

# Start MCP server
python -m naas_abi_core.apps.mcp.mcp_server

# Or with HTTP transport
MCP_TRANSPORT=http python -m naas_abi_core.apps.mcp.mcp_server

Configuration

ABI uses YAML configuration files (typically config.yaml) to configure:

• Services: Storage backends, connection details
• Modules: Which modules to load and their configurations
• API: API title, description, CORS settings
• Global Config: AI mode (cloud, local, airgap)

Example Configuration:

api:
  title: "My ABI API"
  description: "API for my AI system"
  cors_origins:
    - "http://localhost:9879"

global_config:
  ai_mode: "cloud"  # or "local" or "airgap"

services:
  triple_store:
    type: "oxigraph"
    url: "http://localhost:7878"
  
  vector_store:
    type: "qdrant"
    url: "http://localhost:6333"
  
  object_storage:
    type: "minio"
    endpoint: "http://localhost:9000"
  
  secret:
    type: "env"

modules:
  - module: "naas_abi"
    enabled: true
  - path: "./src/my_module"
    enabled: true
    config:
      api_key: "${MY_API_KEY}"

Key Dependencies

• rdflib: RDF and ontology management
• langgraph: Agent conversation management
• langgraph-checkpoint-postgres: Conversation persistence
• fastapi: REST API framework
• sparqlwrapper: SPARQL query execution
• pydantic: Data validation and configuration
• loguru: Logging
• langchain-openai: OpenAI integration

Architecture Patterns

Hexagonal Architecture (Ports & Adapters)

All services follow the Hexagonal Architecture pattern:

• Primary Port: Interface defining service capabilities
• Secondary Adapters: Implementations for different backends
• Benefits: Easy swapping of implementations, testability, system-agnostic design

Module System

• Dependency Resolution: Automatic dependency resolution and ordering
• Lazy Loading: Services loaded only when needed by modules
• Lifecycle Hooks: on_load() and on_initialized() for setup
• Isolation: Modules can be developed and tested independently

Event-Driven Pipelines

Pipelines can be triggered by:

• Manual execution via API or code
• Ontology events (triple insertions/updates)
• Scheduled jobs
• Workflow triggers

Development

Running Tests

pytest

Type Checking

mypy naas_abi_core

Building

uv build

See Also

• ABI Main README - Complete ABI framework documentation
• naas-abi-cli - CLI tool for ABI projects
• naas-abi-marketplace - Marketplace modules and agents

License

MIT License