developer_guide.md

Developer Guide

This guide provides detailed information for developers who want to extend or modify the Agentic RAG system.

Project Structure

The project is organized into several directories:

agentic-rag/
├── agents/            # Agent implementations
├── docs/              # Documentation
├── examples/          # Example usage scripts
├── memory/            # Memory components
├── planning/          # Planning components
├── tests/             # Test suite
├── app.py             # Main application
├── api.py             # API server
├── core.py            # Core data models
├── config.json        # Configuration
├── .env.example       # Environment variables template
├── Dockerfile         # Docker configuration
├── requirements.txt   # Dependencies
└── README.md          # Project overview

Core Components

Data Models

The core data models are defined in core.py:

• Query: Represents a user query to the system
• Document: Represents a document retrieved or generated by the system
• MemoryEntry: Represents an entry in the memory system
• AgentType: Enum for agent types
• AgentMessage: Message exchanged between agents
• AgentResult: Result returned by an agent after processing
• PlanStep: Represents a step in an execution plan
• Plan: Represents an execution plan for a query
• RagOutput: Final output of the Agentic RAG system

Main Application

The main application is defined in app.py. The AgenticRag class orchestrates the workflow between different components.

API Server

The API server is defined in api.py using FastAPI. It provides endpoints for interacting with the system.

Extending the System

Adding a New Agent

To add a new agent:

1. Create a new file in the agents directory (e.g., agents/my_agent.py)
2. Define a class that inherits from BaseAgent
3. Implement the required methods
4. Add configuration in config.json
5. Register the agent in app.py

Example:

from core import AgentType, Query, Document, AgentResult
from agents.base import BaseAgent

class MyAgent(BaseAgent):
    def __init__(self, custom_param: str = "default") -> None:
        super().__init__(agent_type=AgentType.CUSTOM)
        self.custom_param = custom_param
        self.logger.info(f"MyAgent initialized with custom_param={custom_param}")
    
    @BaseAgent.measure_execution_time
    def process(self, query: Query) -> AgentResult:
        self.logger.debug(f"Processing query: {query.id}")
        
        try:
            # Process the query
            content = f"Result for query: {query.text}"
            document = Document(content=content, source="my_agent")
            
            return AgentResult(
                agent_id=self.id,
                agent_type=self.agent_type,
                query_id=query.id,
                documents=[document],
                confidence=0.8,
                processing_time=0.0,  # Will be set by decorator
                metadata={"custom_param": self.custom_param}
            )
        except Exception as e:
            self.logger.error(f"Error processing query: {str(e)}")
            
            return AgentResult(
                agent_id=self.id,
                agent_type=self.agent_type,
                query_id=query.id,
                documents=[],
                confidence=0.0,
                processing_time=0.0,  # Will be set by decorator
                metadata={"error": str(e)}
            )

Adding a New Memory Component

To add a new memory component:

1. Create a new file in the memory directory (e.g., memory/redis_memory.py)
2. Define a class that inherits from BaseMemory
3. Implement the required methods
4. Add configuration in config.json
5. Register the memory component in app.py

Example:

import json
import redis
from typing import Dict, List, Optional, Union

from core import Query, Document, MemoryEntry, AgentResult
from memory.base import BaseMemory

class RedisMemory(BaseMemory):
    def __init__(self, host: str = "localhost", port: int = 6379, db: int = 0) -> None:
        super().__init__()
        self.host = host
        self.port = port
        self.db = db
        self.client = redis.Redis(host=host, port=port, db=db)
        self.logger.info(f"RedisMemory initialized with host={host}, port={port}, db={db}")
    
    def retrieve(self, query: Query) -> Optional[AgentResult]:
        # Implementation
        pass
    
    def store(self, query: Query, result: AgentResult) -> None:
        # Implementation
        pass
    
    def update(self, memory_entry: MemoryEntry) -> None:
        # Implementation
        pass
    
    def remove(self, memory_id: str) -> bool:
        # Implementation
        pass
    
    def clear(self) -> None:
        # Implementation
        pass
    
    def get_stats(self) -> Dict[str, Union[int, float]]:
        # Implementation
        pass
    
    def close(self) -> None:
        self.client.close()
        self.logger.info("Redis connection closed")

Adding a New Planner

To add a new planner:

1. Create a new file in the planning directory (e.g., planning/my_planner.py)
2. Define a class that inherits from BasePlanner
3. Implement the required methods
4. Add configuration in config.json
5. Register the planner in app.py

Example:

from typing import Dict, List, Optional

from core import AgentType, Query, Plan
from planning.base import BasePlanner
from agents.base import BaseAgent

class MyPlanner(BasePlanner):
    def __init__(self, max_depth: int = 3) -> None:
        super().__init__()
        self.max_depth = max_depth
        self.logger.info(f"MyPlanner initialized with max_depth={max_depth}")
    
    def create_plan(self, query: Query, available_agents: Dict[AgentType, BaseAgent]) -> Plan:
        # Implementation
        pass
    
    def adapt_plan(self, plan: Plan, results_so_far: List, available_agents: Dict[AgentType, BaseAgent]) -> Plan:
        # Implementation
        pass

API Integration

To integrate with the API:

1. Use the /query endpoint to process queries
2. Use the /health endpoint to check system health
3. Use the /stats endpoint to get system statistics

Example API request:

import requests
import json

def process_query(query_text, api_url="http://localhost:8000"):
    response = requests.post(
        f"{api_url}/query",
        json={"text": query_text, "metadata": {"source": "api"}}
    )
    
    if response.status_code == 200:
        return response.json()
    else:
        raise Exception(f"Error: {response.status_code} - {response.text}")

result = process_query("What are the latest developments in AI research?")
print(json.dumps(result, indent=2))

Testing

The system includes a comprehensive test suite in the tests directory:

• tests/test_core.py: Tests for core data models
• tests/test_memory.py: Tests for memory components
• tests/test_planning.py: Tests for planning components
• tests/test_agent.py: Tests for agent components
• tests/test_api.py: Tests for API components

To run the tests:

pytest tests/

To run a specific test:

pytest tests/test_core.py::TestCoreModels::test_query_creation

To run tests with coverage:

pytest --cov=. tests/

Logging

The system uses Python's built-in logging module:

import logging

# Get a logger
logger = logging.getLogger("agentic_rag.my_component")

# Log messages
logger.debug("Debug message")
logger.info("Info message")
logger.warning("Warning message")
logger.error("Error message")
logger.critical("Critical message")

Configure logging in config.json:

"logging": {
  "level": "debug",
  "format": "text",
  "output": "both",
  "file_path": "logs/agentic-rag.log"
}

Error Handling

Follow these guidelines for error handling:

1. Use try/except blocks to catch exceptions
2. Log errors with appropriate context
3. Return graceful fallbacks when possible
4. Propagate errors up when necessary

Example:

def process_data(data):
    try:
        # Process data
        result = do_processing(data)
        return result
    except ValueError as e:
        logger.warning(f"Invalid data format: {str(e)}")
        return None
    except Exception as e:
        logger.error(f"Unexpected error processing data: {str(e)}")
        raise

Documentation

Follow these guidelines for documentation:

1. Use docstrings for all modules, classes, and functions
2. Follow the Google Python Style for docstrings
3. Include type annotations
4. Keep documentation up to date with code changes

Example:

def process_query(query_text: str) -> Dict[str, Any]:
    """
    Process a query through the Agentic RAG system.
    
    Args:
        query_text: Text of the query to process
        
    Returns:
        Dictionary containing the response and metadata
        
    Raises:
        ValueError: If query_text is empty
    """
    if not query_text:
        raise ValueError("Query text cannot be empty")
    
    # Process query
    return {"response": "Result", "metadata": {}}

Performance Considerations

1. Use the @BaseAgent.measure_execution_time decorator to track performance
2. Monitor memory usage, especially in memory components
3. Use caching where appropriate
4. Consider asynchronous processing for I/O-bound operations
5. Optimize database queries and indexing

Security Considerations

1. Validate all user input
2. Use environment variables for sensitive configuration
3. Implement proper authentication for API endpoints
4. Keep dependencies up to date
5. Follow the principle of least privilege
6. Implement rate limiting and other protection measures

Deployment

Deploy the system using Docker:

# Build the Docker image
docker build -t agentic-rag .

# Run the container
docker run -p 8000:8000 \
  -v $(pwd)/config.json:/app/config.json \
  -v $(pwd)/data:/app/data \
  -v $(pwd)/logs:/app/logs \
  --env-file .env \
  agentic-rag

For production environments:

1. Use a reverse proxy (e.g., Nginx, Traefik)
2. Implement proper authentication
3. Set up monitoring and alerting
4. Configure auto-scaling
5. Use a container orchestration system (e.g., Kubernetes, Docker Swarm)

Continuous Integration/Deployment

Set up CI/CD pipelines using GitHub Actions, GitLab CI, or Jenkins:

1. Run linting and static analysis
2. Run unit and integration tests
3. Build Docker image
4. Push to container registry
5. Deploy to staging environment
6. Run acceptance tests
7. Deploy to production environment

For more information, see the Operations Guide.