Developing VAgents Packages

Developing VAgents Packages

VAgents is a powerful framework for building, packaging, and distributing AI agents with sophisticated capabilities. This guide covers creating VAgents packages that leverage the framework’s advanced features including agent orchestration, tool integration, and intelligent workflow management.

Quick Start

Create a new VAgents package using the built-in template generator:

vibe create-template my-agent-package

Or specify a custom directory:

vibe create-template my-agent-package --output-dir ./agents

This generates a production-ready VAgents package structure with intelligent defaults.

VAgents Package Architecture

VAgents packages follow a structured approach optimized for agent development:

my-agent-package/
├── package.yaml              # VAgents configuration with agent capabilities
├── my_agent_package.py       # Agent implementation with VAgents integration
├── README.md                 # Documentation and usage examples
└── requirements.txt          # Python dependencies (VAgents included)

Package Components

VAgents packages include:

1. Agent Configuration - Define agent capabilities, tools, and workflows
2. VAgents Integration - Leverage the VAgents framework’s powerful features
3. Tool Definitions - Extensible tool system for agent capabilities
4. Workflow Management - Intelligent task orchestration and execution

VAgents Configuration

Agent Definition

name: intelligent-data-processor
version: 1.0.0
description: An intelligent agent for data processing and analysis
author: VAgents Developer
repository_url: https://github.com/vagents-ai/intelligent-data-processor.git
entry_point: intelligent_data_processor.main
dependencies:
  - vagents>=2.0.0
  - pandas
  - numpy
python_version: ">=3.9"
tags:
  - ai-agent
  - data-processing
  - intelligent-automation

Agent Capabilities Configuration

Define your agent’s capabilities and interface using VAgents’ advanced argument system:

arguments:
  # Agent mode selection
  - name: mode
    type: str
    help: Agent execution mode
    short: m
    choices: [analyze, process, interactive]
    default: analyze
    required: false

  # Data source configuration
  - name: source
    type: str
    help: Data source (file, url, or stdin)
    short: s
    required: false

  # Agent verbosity
  - name: verbose
    type: bool
    help: Enable detailed agent logging
    short: v
    required: false

  # Tool selection
  - name: tools
    type: list
    help: Available tools for the agent
    short: t
    required: false

  # Configuration profile
  - name: profile
    type: str
    help: Agent configuration profile
    choices: [development, production, testing]
    default: production
    required: false

Advanced Agent Configuration

arguments:
  # Model configuration
  - name: model
    type: str
    help: AI model to use for agent reasoning
    default: gpt-4
    required: false

  # Concurrency settings
  - name: max-concurrent
    type: int
    help: Maximum concurrent agent operations
    default: 4
    required: false

  # Performance tuning
  - name: timeout
    type: float
    help: Agent operation timeout in seconds
    default: 30.0
    required: false

  # Agent memory
  - name: memory-size
    type: int
    help: Agent memory buffer size
    default: 1000
    required: false

VAgents Implementation Patterns

AgentModule + Protocols (recommended)

Packages can expose an agentified entry point using AgentModule and the AgentInput/AgentOutput protocol. The package manager detects these automatically and will wrap plain dict outputs into AgentOutput.

Class-based entry point:

my_agent_package.py

from vagents.core import AgentModule, AgentInput, AgentOutput

class MyAgent(AgentModule):
    async def forward(self, input: AgentInput) -> AgentOutput:
        # do work here
        return AgentOutput(input_id=input.id, result={"echo": input.payload})

Function-based entry point:

my_agent_package.py

from vagents.core import AgentInput, AgentOutput

async def run(input: AgentInput) -> AgentOutput:
    return AgentOutput(input_id=input.id, result={"echo": input.payload})

In your package.yaml set:

entry_point: my_agent_package.MyAgent   # or my_agent_package.run

Basic VAgents Integration

from vagents import VAgent, Tool, Workflow
from typing import Dict, Any, Optional

def main(
    mode: str = "analyze",
    source: Optional[str] = None,
    verbose: bool = False,
    tools: Optional[list] = None,
    input: Optional[str] = None,
    stdin: Optional[str] = None,
    **kwargs
) -> Dict[str, Any]:
    """
    VAgents-powered intelligent data processor

    Args:
        mode: Agent execution mode (analyze, process, interactive)
        source: Data source specification
        verbose: Enable detailed agent logging
        tools: List of available tools
        input: Input content from stdin
        stdin: Standard input content (alias for input)
        **kwargs: Additional VAgents configuration

    Returns:
        dict: Agent execution results with VAgents metadata
    """
    # Initialize VAgents framework
    agent = VAgent(
        name="intelligent-data-processor",
        mode=mode,
        verbose=verbose,
        tools=tools or ["data_analyzer", "file_processor", "pattern_detector"]
    )

    # Handle input data
    content = input or stdin
    if source:
        content = agent.load_data(source)

    # Execute agent workflow
    result = {
        "agent_name": agent.name,
        "mode": mode,
        "status": "initialized",
        "tools_available": agent.get_available_tools(),
        "has_input": content is not None
    }

    if content:
        # Run intelligent processing
        processed = agent.process(content, mode=mode)
        result.update({
            "status": "completed",
            "processed_data": processed,
            "execution_metadata": agent.get_execution_metadata()
        })

        if verbose:
            print(f"VAgent processed {len(content)} characters")
            print(f"Used tools: {agent.get_used_tools()}")

    return result

class DataAnalyzer(Tool):
    """VAgents tool for intelligent data analysis"""

    def __init__(self):
        super().__init__(name="data_analyzer", description="Analyze data patterns and insights")

    def execute(self, data: str, **kwargs) -> Dict[str, Any]:
        """Execute data analysis using VAgents capabilities"""
        return {
            "analysis_type": "pattern_recognition",
            "insights": self._extract_insights(data),
            "recommendations": self._generate_recommendations(data)
        }

    def _extract_insights(self, data: str) -> list:
        """Extract intelligent insights from data"""
        # VAgents-powered analysis logic
        return ["insight1", "insight2", "insight3"]

    def _generate_recommendations(self, data: str) -> list:
        """Generate actionable recommendations"""
        # VAgents recommendation engine
        return ["recommendation1", "recommendation2"]

Advanced VAgents Agent Implementation

from vagents import VAgent, Workflow, ToolRegistry, AgentMemory
from vagents.core import Session, LLM
import json
from pathlib import Path
from typing import Dict, Any, Optional

class IntelligentDataProcessor(VAgent):
    """Advanced VAgents implementation with full framework integration"""

    def __init__(self, **config):
        super().__init__(
            name="intelligent-data-processor",
            description="AI-powered data processing agent",
            **config
        )

        # Initialize VAgents components
        self.session = Session()
        self.memory = AgentMemory(size=config.get("memory_size", 1000))
        self.tool_registry = ToolRegistry()
        self.workflow = Workflow()

        # Register tools
        self._register_tools()

    def _register_tools(self):
        """Register VAgents tools"""
        self.tool_registry.register([
            DataAnalyzer(),
            PatternDetector(),
            ReportGenerator(),
            FileProcessor()
        ])

    def execute(
        self,
        mode: str = "analyze",
        model: str = "gpt-4",
        max_concurrent: int = 4,
        timeout: float = 30.0,
        input: Optional[str] = None,
        stdin: Optional[str] = None,
        **kwargs
    ) -> Dict[str, Any]:
        """Execute agent with VAgents framework"""

        try:
            # Configure VAgents session
            self.session.configure(
                model=model,
                max_concurrent=max_concurrent,
                timeout=timeout
            )

            content = input or stdin

            # Create VAgents workflow
            workflow_result = self.workflow.execute([
                {"tool": "data_analyzer", "input": content},
                {"tool": "pattern_detector", "input": content},
                {"tool": "report_generator", "dependencies": ["data_analyzer", "pattern_detector"]}
            ])

            # Store in agent memory
            self.memory.store(workflow_result)

            return {
                "agent": self.name,
                "framework": "VAgents",
                "mode": mode,
                "status": "success",
                "workflow_result": workflow_result,
                "session_metadata": self.session.get_metadata(),
                "memory_usage": self.memory.get_usage_stats(),
                "tools_executed": self.workflow.get_executed_tools()
            }

        except Exception as e:
            return {
                "agent": self.name,
                "framework": "VAgents",
                "status": "error",
                "error": str(e),
                "error_type": type(e).__name__,
                "session_state": self.session.get_state()
            }

# VAgents entry point
main = IntelligentDataProcessor()

VAgents Development Workflow

1. Initialize VAgents Project

# Create VAgents package template
vibe create-template intelligent-processor --output-dir ./agents

# Navigate to agent directory
cd agents/intelligent-processor

# Initialize VAgents development environment
git init
git add .
git commit -m "Initial VAgents agent structure"

2. VAgents Development and Testing

# Test VAgents integration directly
python intelligent_processor.py --mode analyze --verbose

# Test agent with data input
echo "sample data" | python intelligent_processor.py --mode process --tools data_analyzer

# Test VAgents workflow
cat dataset.json | python intelligent_processor.py --mode interactive --model gpt-4

3. VAgents Local Integration

# Install VAgents package locally
vibe install /absolute/path/to/agents/intelligent-processor

# Test via VAgents package manager
vibe run intelligent-processor --mode analyze --verbose

# Test VAgents workflows
cat data.txt | vibe run intelligent-processor --mode process --max-concurrent 8

4. VAgents Publishing

# Prepare VAgents package for distribution
# Update package.yaml with VAgents-specific metadata
git remote add origin https://github.com/vagents-ai/intelligent-processor.git
git push -u origin main

# Tag VAgents release
git tag v1.0.0
git push --tags

5. VAgents Distribution

# Users install VAgents packages with
vibe install https://github.com/vagents-ai/intelligent-processor.git

# Or specific VAgents version
vibe install https://github.com/vagents-ai/intelligent-processor.git --branch v1.0.0

VAgents Testing and Validation

VAgents Agent Testing

# Test VAgents agent capabilities
vibe run intelligent-processor --mode analyze --tools data_analyzer pattern_detector --verbose

# Test VAgents workflow execution
echo "complex data" | vibe run intelligent-processor --mode interactive --model gpt-4
cat large_dataset.json | vibe run intelligent-processor --mode process --max-concurrent 4

# Test VAgents error handling
vibe run intelligent-processor --mode invalid  # Should fail gracefully with VAgents error handling
vibe run intelligent-processor --timeout 0.1  # Should handle timeout appropriately

VAgents Integration Testing

Create comprehensive tests for your VAgents package:

test_vagents_integration.py

import subprocess
import json
from vagents.testing import VAgentsTestSuite

class TestIntelligentProcessor(VAgentsTestSuite):
    """Test suite for VAgents integration"""

    def test_vagents_initialization(self):
        """Test VAgents framework initialization"""
        result = subprocess.run(
            ["vibe", "run", "intelligent-processor", "--mode", "analyze"],
            capture_output=True, text=True
        )
        assert result.returncode == 0
        data = json.loads(result.stdout)
        assert data["framework"] == "VAgents"
        assert "agent" in data

    def test_vagents_workflow(self):
        """Test VAgents workflow execution"""
        process = subprocess.Popen(
            ["echo", "test data for VAgents"],
            stdout=subprocess.PIPE
        )
        result = subprocess.run(
            ["vibe", "run", "intelligent-processor", "--mode", "process", "--verbose"],
            stdin=process.stdout, capture_output=True, text=True
        )
        assert result.returncode == 0
        assert "VAgent processed" in result.stdout
        assert "tools_executed" in result.stdout

    def test_vagents_tools_integration(self):
        """Test VAgents tools system"""
        result = subprocess.run(
            ["vibe", "run", "intelligent-processor", "--tools", "data_analyzer", "pattern_detector"],
            capture_output=True, text=True
        )
        assert result.returncode == 0
        data = json.loads(result.stdout)
        assert "tools_available" in data
        assert len(data["tools_available"]) == 2

if __name__ == "__main__":
    suite = TestIntelligentProcessor()
    suite.run_all_tests()
    print("All VAgents integration tests passed!")

VAgents Best Practices

1. VAgents Framework Integration

• Leverage VAgents’ built-in agent orchestration capabilities
• Use VAgents tools system for extensible functionality
• Implement proper VAgents session management
• Follow VAgents workflow patterns for complex operations

2. Intelligent Agent Design

• Design agents with clear capabilities and boundaries
• Implement robust error handling with VAgents error management
• Use VAgents memory system for context retention
• Leverage VAgents’ LLM integration for intelligent decision-making

3. VAgents Tool Development

• Create reusable tools using VAgents Tool base class
• Register tools properly with VAgents ToolRegistry
• Implement tool validation and error handling
• Design tools for composability and chaining

4. Performance and Scalability

• Use VAgents’ concurrent execution capabilities
• Implement proper timeout and resource management
• Leverage VAgents’ optimization features
• Monitor agent performance using VAgents metrics

5. VAgents Configuration Management

• Use VAgents configuration system for agent settings
• Implement environment-specific configurations
• Validate configurations using VAgents validators
• Support runtime configuration updates

For comprehensive VAgents usage patterns and advanced features, see the VAgents Framework Documentation.