agenticstar-platform 0.3.0

AGENTICSTAR Platform SDK - Enterprise AI Agent Infrastructure

AGENTICSTAR Platform SDK

Enterprise AI Agent Infrastructure SDK for building autonomous agent systems.

Installation

pip install agenticstar-platform

Quick Start

from agenticstar_platform import (
    # Database
    PostgreSQLManager, PostgreSQLConfig,
    # RAG (Vector DB + Embedding)
    QdrantManager, QdrantConfig, EmbeddingGenerator, EmbeddingConfig,
    # Storage
    AzureBlobStorageClient, AzureBlobConfig,
    # Events
    EventEmitter, EventType,
    # Memory
    SemanticMemoryClient, SemanticMemoryConfig,
)

# Example: Initialize SDK components
async def main():
    # Database
    db_config = PostgreSQLConfig.from_toml("config.toml", section="database")
    async with PostgreSQLManager(db_config) as db:
        users = await db.fetch_all("SELECT * FROM users WHERE active = $1", (True,))

    # RAG System
    embedding_config = EmbeddingConfig.from_toml("config.toml", section="rag.embedding")
    embedding_gen = EmbeddingGenerator(embedding_config)

    qdrant_config = QdrantConfig.from_toml("config.toml", section="rag.qdrant")
    async with QdrantManager(qdrant_config, embedding_gen) as qdrant:
        results = await qdrant.search("How to use the SDK?", limit=5)

Modules

• db: PostgreSQL data access layer with Azure AD support
• rag: Qdrant vector database and Azure OpenAI embedding integration
• storage: Multi-cloud storage (Azure Blob, S3, GCS)
• security: PII detection (Azure Presidio, AWS Comprehend, GCP DLP)
• memory: Episodic (Graphiti/FalkorDB) and semantic (Mem0) memory
• events: Event type definitions for streaming
• common: Shared utilities (secret masking, validation)

Platform Class Example

Below is an example of a Platform class that wraps SDK components for your agent system:

"""
Platform class example - Using AGENTICSTAR Platform SDK
"""
import asyncio
from dataclasses import dataclass
from typing import Optional

from agenticstar_platform import (
    PostgreSQLManager, PostgreSQLConfig,
    QdrantManager, QdrantConfig,
    EmbeddingGenerator, EmbeddingConfig,
    EventEmitter, EventType,
    SemanticMemoryClient, SemanticMemoryConfig,
    AzureBlobStorageClient, AzureBlobConfig,
)


@dataclass
class PlatformConfig:
    """Platform configuration"""
    db_config: PostgreSQLConfig
    qdrant_config: QdrantConfig
    embedding_config: EmbeddingConfig
    storage_config: Optional[AzureBlobConfig] = None
    memory_config: Optional[SemanticMemoryConfig] = None

    @classmethod
    def from_toml(cls, path: str) -> "PlatformConfig":
        """Load all configurations from TOML file"""
        return cls(
            db_config=PostgreSQLConfig.from_toml(path, section="database"),
            qdrant_config=QdrantConfig.from_toml(path, section="rag.qdrant"),
            embedding_config=EmbeddingConfig.from_toml(path, section="rag.embedding"),
            storage_config=AzureBlobConfig.from_dict({
                # Load from environment or config
                "bucket_name": "your-container",
                "connection_string": "your-connection-string",
            }),
        )


class AgentPlatform:
    """
    Platform class wrapping SDK components.

    This class demonstrates how to use AGENTICSTAR Platform SDK
    to build your own agent infrastructure.

    Example:
        >>> config = PlatformConfig.from_toml("config.toml")
        >>> platform = AgentPlatform(config)
        >>> await platform.initialize()
        >>>
        >>> # Use database
        >>> users = await platform.db.fetch_all("SELECT * FROM users")
        >>>
        >>> # Use RAG
        >>> results = await platform.search_knowledge("How to deploy?")
        >>>
        >>> # Clean up
        >>> await platform.cleanup()
    """

    def __init__(self, config: PlatformConfig):
        self.config = config
        self._db: Optional[PostgreSQLManager] = None
        self._qdrant: Optional[QdrantManager] = None
        self._embedding: Optional[EmbeddingGenerator] = None
        self._storage: Optional[AzureBlobStorageClient] = None
        self._memory: Optional[SemanticMemoryClient] = None

    async def initialize(self) -> None:
        """Initialize all SDK components"""
        # Database
        self._db = PostgreSQLManager(self.config.db_config)
        await self._db.initialize()

        # Embedding generator
        self._embedding = EmbeddingGenerator(self.config.embedding_config)

        # Vector DB (RAG)
        self._qdrant = QdrantManager(self.config.qdrant_config, self._embedding)
        await self._qdrant.initialize()

        # Storage (optional)
        if self.config.storage_config:
            self._storage = AzureBlobStorageClient(self.config.storage_config)

        # Memory (optional)
        if self.config.memory_config:
            self._memory = SemanticMemoryClient(self.config.memory_config)

    @property
    def db(self) -> PostgreSQLManager:
        """Database manager"""
        if not self._db:
            raise RuntimeError("Platform not initialized. Call initialize() first.")
        return self._db

    @property
    def qdrant(self) -> QdrantManager:
        """Qdrant manager"""
        if not self._qdrant:
            raise RuntimeError("Platform not initialized. Call initialize() first.")
        return self._qdrant

    @property
    def storage(self) -> Optional[AzureBlobStorageClient]:
        """Storage client (optional)"""
        return self._storage

    @property
    def memory(self) -> Optional[SemanticMemoryClient]:
        """Memory client (optional)"""
        return self._memory

    async def search_knowledge(self, query: str, limit: int = 10):
        """
        Search knowledge base using RAG

        Args:
            query: Search query
            limit: Max results

        Returns:
            Search results from vector DB
        """
        return await self.qdrant.search(query, limit=limit)

    async def add_knowledge(
        self,
        point_id: str,
        text: str,
        metadata: dict,
    ) -> dict:
        """
        Add document to knowledge base

        Args:
            point_id: Document ID
            text: Document content
            metadata: Document metadata

        Returns:
            Upsert result
        """
        return await self.qdrant.upsert(
            point_id=point_id,
            text=text,
            payload=metadata,
        )

    async def upload_file(self, file_path: str, prefix: str = "") -> dict:
        """
        Upload file to storage

        Args:
            file_path: Local file path
            prefix: Storage prefix (folder)

        Returns:
            Upload result
        """
        if not self._storage:
            return {"success": False, "error": "Storage not configured"}

        result = await self._storage.upload_file(file_path, prefix=prefix)
        return {
            "success": result.success,
            "url": result.object_url,
            "size": result.file_size,
        }

    async def cleanup(self) -> None:
        """Clean up all resources"""
        if self._qdrant:
            await self._qdrant.close()
        if self._db:
            await self._db.close()
        if self._storage:
            await self._storage.close()
        if self._memory:
            await self._memory.cleanup()


# Usage example
async def example_usage():
    # Load configuration
    config = PlatformConfig.from_toml("config.toml")

    # Initialize platform
    platform = AgentPlatform(config)
    await platform.initialize()

    try:
        # Add knowledge
        await platform.add_knowledge(
            point_id="doc-001",
            text="AGENTICSTAR Platform SDKはエンタープライズAIエージェント基盤のためのSDKです。",
            metadata={"category": "documentation", "language": "ja"},
        )

        # Search knowledge
        results = await platform.search_knowledge("SDKの使い方")
        for result in results.get("results", []):
            print(f"Score: {result['score']:.2f} - {result['text'][:50]}...")

        # Database query
        users = await platform.db.fetch_all(
            "SELECT id, name FROM users WHERE created_at > $1",
            (datetime(2024, 1, 1),)
        )
        print(f"Found {len(users)} users")

    finally:
        await platform.cleanup()


if __name__ == "__main__":
    asyncio.run(example_usage())

Configuration (config.toml example)

[database]
host = "your-postgresql.postgres.database.azure.com"
port = 5432
database = "agenticai"
username = "admin"
password = "your-password"
use_azure_ad = false
pool_min_size = 2
pool_max_size = 10

[database.azure_ad]
tenant_id = "your-tenant-id"
client_id = "your-client-id"
client_secret = "your-client-secret"

[rag.embedding]
api_base = "https://your-openai.openai.azure.com/"
api_key = "your-api-key"
deployment_name = "text-embedding-ada-002"

[rag.qdrant]
url = "http://localhost:6333"
collection_name = "knowledge_base"
vector_size = 1536

[storage.azure]
bucket_name = "your-container"
connection_string = "DefaultEndpointsProtocol=https;..."
prefix = "uploads/"

[memory.llm]
model = "azure/gpt-4"
api_key = "your-api-key"
base_url = "https://your-openai.openai.azure.com/"
api_version = "2024-02-15-preview"

[memory.embedder]
model = "azure/text-embedding-ada-002"
api_key = "your-api-key"
base_url = "https://your-openai.openai.azure.com/"
api_version = "2024-02-15-preview"

API Reference

See API_REFERENCE.md for detailed API documentation.

Version

0.1.0