Development Guide

Development Environment Setup

Prerequisites

• Python 3.11+

• Poetry for dependency management

• Docker (optional)

• Git

Local Development Setup

1. Clone the repository:

git clone https://github.com/0xdevpro/deepcore.git
cd deepcore

1. Install dependencies:

poetry install

1. Set up environment variables:

cp .env.example .env

1. Configure your .env file:

HOST=localhost
PORT=8000
DEBUG=true
JWT_SECRET=your_jwt_secret
DATABASE_URL=postgresql://user:password@localhost:5432/deepcore

1. Start the development server:

poetry run python api.py

Project Structure

deepcore/
├── agents/
│   ├── agent/          # Core agent implementation
│   ├── api/            # API endpoints
│   ├── common/         # Shared utilities
│   ├── middleware/     # Middleware components
│   ├── models/         # Data models
│   ├── protocol/       # Protocol definitions
│   ├── services/       # Business logic
│   └── utils/          # Utility functions
├── sql/                # Database migrations
├── tests/              # Test suite
├── api.py             # Main application entry
├── pyproject.toml     # Project dependencies
└── README.md          # Project documentation

Core Components

Agent System

The agent system is the heart of DeepCore. It consists of several key components:

1. Agent Core

from agents.agent.base import BaseAgent
​
class CustomAgent(BaseAgent):
    def __init__(self, config: dict):
        super().__init__(config)
        self.tools = []
        self.memory = []
​
    async def process(self, input_data: str) -> str:
        # Implement agent processing logic
        response = await self._think(input_data)
        return response
​
    async def _think(self, input_data: str) -> str:
        # Implement thinking logic
        pass

1. Tool Integration

from agents.agent.tool import BaseTool
​
class CustomTool(BaseTool):
    def __init__(self):
        super().__init__(
            name="custom_tool",
            description="Tool description"
        )
​
    async def _run(self, input_data: dict) -> dict:
        # Implement tool logic
        result = await self.process_data(input_data)
        return {"result": result}
​
    async def process_data(self, data: dict) -> str:
        # Implement specific data processing
        pass

API Layer

The API layer is built using FastAPI:

from fastapi import APIRouter, Depends
from agents.services.agent_service import AgentService

router = APIRouter()

@router.post("/agent/create")
async def create_agent(
    data: AgentCreate,
    service: AgentService = Depends()
):
    return await service.create_agent(data)

@router.get("/agent/{agent_id}")
async def get_agent(
    agent_id: str,
    service: AgentService = Depends()
):
    return await service.get_agent(agent_id)

Service Layer

Services handle business logic:

from agents.models.db import AgentModel
from agents.protocol.schemas import AgentCreate

class AgentService:
    async def create_agent(self, data: AgentCreate) -> AgentModel:
        # Validate data
        self._validate_agent_data(data)
        
        # Create agent
        agent = await AgentModel.create(**data.dict())
        
        # Initialize tools
        await self._initialize_tools(agent)
        
        return agent

    async def _initialize_tools(self, agent: AgentModel):
        # Tool initialization logic
        pass

Development Guidelines

Code Style

1. Follow PEP 8 guidelines

2. Use type hints

3. Document functions and classes

4. Write meaningful commit messages

Example:

from typing import Optional, List

class DataProcessor:
    """
    Processes data for agent consumption.
    
    Attributes:
        batch_size: Size of data batches to process
        max_items: Maximum number of items to process
    """
    
    def __init__(self, batch_size: int = 100, max_items: Optional[int] = None):
        self.batch_size = batch_size
        self.max_items = max_items
    
    def process_batch(self, items: List[dict]) -> List[dict]:
        """
        Process a batch of items.
        
        Args:
            items: List of items to process
            
        Returns:
            List of processed items
        """
        return [self._process_item(item) for item in items]

Testing

1. Write unit tests for all components

2. Use pytest for testing

3. Maintain high test coverage

Example:

import pytest
from agents.services.agent_service import AgentService

@pytest.fixture
def agent_service():
    return AgentService()

@pytest.mark.asyncio
async def test_agent_creation(agent_service):
    data = AgentCreate(
        name="test_agent",
        description="Test agent",
        mode="ReAct"
    )
    
    agent = await agent_service.create_agent(data)
    assert agent.name == "test_agent"
    assert agent.mode == "ReAct"

Error Handling

Implement proper error handling:

from agents.exceptions import AgentError

class AgentService:
    async def create_agent(self, data: AgentCreate) -> AgentModel:
        try:
            # Attempt to create agent
            agent = await AgentModel.create(**data.dict())
            return agent
        except Exception as e:
            raise AgentError(f"Failed to create agent: {str(e)}")

Database Management

Migrations

Use Alembic for database migrations:

# Create a new migration
alembic revision --autogenerate -m "Add agent table"

# Run migrations
alembic upgrade head

Models

Define SQLAlchemy models:

from sqlalchemy import Column, String, Integer
from agents.models.base import Base

class Agent(Base):
    __tablename__ = "agents"
    
    id = Column(String, primary_key=True)
    name = Column(String, nullable=False)
    description = Column(String)
    mode = Column(String, nullable=False)

Deployment

Docker Deployment

1. Build the image:

docker build -t deepcore .

1. Run the container:

docker run -p 8000:8000 \
    -e DATABASE_URL=postgresql://user:password@host:5432/deepcore \
    -e JWT_SECRET=your_secret \
    deepcore

Production Considerations

1. Use proper logging:

import logging

logger = logging.getLogger(__name__)

class AgentService:
    async def create_agent(self, data: AgentCreate):
        logger.info(f"Creating agent: {data.name}")
        try:
            agent = await AgentModel.create(**data.dict())
            logger.info(f"Agent created successfully: {agent.id}")
            return agent
        except Exception as e:
            logger.error(f"Failed to create agent: {str(e)}")
            raise

1. Implement monitoring:

from prometheus_client import Counter, Histogram

request_count = Counter(
    'agent_requests_total',
    'Total agent requests'
)

request_latency = Histogram(
    'agent_request_latency_seconds',
    'Agent request latency'
)

Contributing

1. Fork the repository

2. Create a feature branch

3. Write tests

4. Submit a pull request

Pull Request Guidelines

• Follow the code style guide

• Include tests

• Update documentation

• Add changelog entry

Troubleshooting

Common Issues

1. Database Connection Issues:

try:
    await database.connect()
except Exception as e:
    logger.error(f"Database connection failed: {str(e)}")
    sys.exit(1)

1. Authentication Problems:

if token_error:
    logger.warning(f"Authentication failed for user: {user_id}")
    raise AuthenticationError("Invalid token")

Debugging

Enable debug logging:

import logging

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

logger.debug("Detailed debugging information")

Performance Optimization

1. Use connection pooling:

from databases import Database

database = Database(
    DATABASE_URL,
    min_size=5,
    max_size=20
)

1. Implement caching:

from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend

@router.get("/agent/{agent_id}")
@cache(expire=60)
async def get_agent(agent_id: str):
    return await agent_service.get_agent(agent_id)

Security Best Practices

1. Input validation:

from pydantic import BaseModel, validator

class UserInput(BaseModel):
    name: str
    email: str
    
    @validator('email')
    def validate_email(cls, v):
        if not '@' in v:
            raise ValueError('Invalid email')
        return v

1. Rate limiting:

from fastapi import Depends
from agents.middleware.rate_limit import rate_limit
​
@router.post("/agent/create")
@rate_limit(max_requests=10, window_seconds=60)
async def create_agent(data: AgentCreate):
    return await agent_service.create_agent(data)