Easy-to-use Python interface for AIHorde Server
# Configuration file for the Sphinx documentation builder.
#
# For the full list of built-in configuration values, see the documentation:
# https://www.sphinx-doc.org/en/master/usage/configuration.html

# -- Project information -----------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#project-information
from datetime import datetime

project = 'horde-client'
copyright = f'{datetime.now().year}, Rahul D Shetty'
author = 'Rahul D Shetty'

source_suffix = ".md"
master_doc = "index"


# -- General configuration ---------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration

extensions = [
    "myst_parser"
]

myst_enable_extensions = [
    "amsmath",
    "attrs_inline",
    "colon_fence",
    "deflist",
    "dollarmath",
    "fieldlist",
    "html_admonition",
    "html_image",
    "linkify",
    "replacements",
    "smartquotes",
    "strikethrough",
    "substitution",
    "tasklist",
]

templates_path = ['_templates']
exclude_patterns = ["_build"]


# -- Options for HTML output -------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output

html_theme = 'alabaster'
html_static_path = ['_static']

# Theme based options
# https://alabaster.readthedocs.io/en/latest/customization.html
html_theme_options = {
    "description": "Easy-to-use Python Interface for KoboldAI Horde",
    'github_button': True,
    'github_user': 'rahuldshetty',
    'github_repo': 'horde-client',


    "fixed_sidebar": True,
    'sidebar_width': '220px',

    'base_text': '#3E4349',
    'body_text': '#3E4349',
    'font_family': 'Georgia, serif',
    'font_size': '17px',
    'head_font_family': 'Georgia, serif',
}
docs/source/conf.py
rahuldshetty/horde-client
import asyncio

from horde_client import AsyncHordeClient, ModelType, TextGenParams

client = AsyncHordeClient(
    insecure=True
)

prompt =  """### Instruction: 
Tell me a knock knock joke.

### Response:
"""

# Text Generation
async def main():
    # List available models
    print("===== Model List =====")
    models = await client.list_models(type=ModelType.text)
    for model in models[:5]:
        print(model)

    # Text Generation
    print("\n===== Prompt =====")
    print(prompt)

    params = TextGenParams(
        max_context_length = 512,
        temperature=0.8
    )

    text_gen_ouput = await client.text_gen(prompt, params=params)

    print("\n===== Output =====")
    print(text_gen_ouput)
    print(text_gen_ouput.generations[0].text)


asyncio.get_event_loop().run_until_complete(main())
examples/async_text_generation.py
rahuldshetty/horde-client
import io, base64
from PIL import Image

from horde_client import HordeClient, ModelType, ImageGenParams

client = HordeClient(
    # To access public KoboldAI service
    insecure=True 
)

# List available models
print("===== Model List =====")
for model in client.list_models(type=ModelType.text)[:20]:
    print(model)

# Set model
client.clear_model()
client.set_model('Project Unreal Engine 5')

# Image Generation Prompt
prompt =  """an orange cat reading newspaper in bench and sitting in park"""

print("\n===== Prompt =====")
print(prompt)

params = ImageGenParams()

image_gen_ouput = client.image_gen(prompt, params=params)

print("\n===== Output =====")
file_output = "my-cat.png"
img = Image.open(io.BytesIO(base64.decodebytes(bytes(image_gen_ouput.generations[0].img, "utf-8"))))
img.save(file_output)
print('Output saved to ' + file_output)

examples/image/image_generation.py
rahuldshetty/horde-client
from horde_client import HordeClientLLM, TextGenParams

from langchain import LLMChain
from langchain.prompts import PromptTemplate

# Prepare HordeClientLLM
params = TextGenParams(
    max_context_length=256,
    max_length=64,
    temperature=0.8
)

llm = HordeClientLLM(
    # To access public Horde Client
    insecure=True,

    # TextGen Parameters for HordeAI Client
    params=params
)

# Prompt Template
template = """### Instruction:
Create a fancy company name for a company that makes {product}.
### Response:
"""

prompt= PromptTemplate(input_variables=["product"], template=template)

# Chain Prompt with LLM
chain = LLMChain(
    llm = llm,
    prompt = prompt
)

print(chain.run("colorful socks"))

print(chain.run("mobiles"))


examples/langchain_demo.py
rahuldshetty/horde-client
from horde_client import HordeClient, ModelType, TextGenParams

client = HordeClient(
    # To access public KoboldAI service
    insecure=True 
)

# List available models
print("===== Model List =====")
for model in client.list_models(type=ModelType.text)[:5]:
    print(model)

# Text Generation
prompt =  """### Instruction: 
Tell me a knock knock joke.

### Response:
"""

print("\n===== Prompt =====")
print(prompt)

params = TextGenParams(
    max_context_length = 512,
    temperature=0.8
)

text_gen_ouput = client.text_gen(prompt, params=params)

print("\n===== Output =====")
print(text_gen_ouput)
print(text_gen_ouput.generations[0].text)

examples/text_generation.py
rahuldshetty/horde-client
from horde_client.client import HordeClient
from horde_client.async_client import AsyncHordeClient

from horde_client.model import (
    ModelType,
    TextGenParams,
    ImageGenParams
)

from horde_client.langchain_llm import HordeClientLLM
horde_client/__init__.py
rahuldshetty/horde-client
import random, time, json
from urllib.parse import  urljoin
from typing import List

import asyncio
from asyncio_requests.asyncio_request import request

from horde_client import config
from horde_client import model
from horde_client import errors

class AsyncHordeClient:
    def __init__(self, 
            base_url:str = config.API_BASE_URL,
            api_token:str = config.ANON_KEY,
            insecure:bool = False
        ):
        '''
        Instantiate Async HordeClient SDK.

        Parameters:
        -----------
        base_url (str):  KoboldAI API URL (Default: https://horde.koboldai.net/api/)
        api_token (str): KoboldAI API Key (Default: 0000000000)
        insecure (bool): Enabling insecure mode allows access to public KoboldAI. 
                         You will need to provide base_url when this flag is off (Default False)
        '''
        self.__base_url = base_url.strip()
        self.__api_token = api_token.strip()
        self.__insecure = insecure
        self.__models = []

        self.validate_insecure_permission()
    
    def validate_insecure_permission(self):
        '''
        To avoid accidental data leaks to public KoboldAI services, users have to manually enabled insecure mode.
        This methods is called during class initialization to validate the checks.
        '''
        if self.__insecure == False and self.__base_url == config.API_BASE_URL:
            raise errors.InsecureModePermissionError()

    def __headers(self):
        '''
        '''
        return {
            'Apikey': self.__api_token,
            'User-Agent': random.choice(config.user_agents)
        }
    
    def __gen_query_params_string(self, query_params: dict):
        if not query_params: return ""
        return "?" + "&".join([ k + "=" + v for k, v in  query_params.items()])

    async def get(self, endpoint, path_params:dict = None, query_params:dict = None):
        '''
        Get Request Endgine
        '''
        endpoint = endpoint
        
        # Format path parameter in endpoint url 
        if path_params:
            endpoint = endpoint.format(
                **path_params
            )
        
        url = urljoin(self.__base_url, endpoint)

        # Additional Header tokens added here
        headers = self.__headers()
        
        # Concatenate Qury Params
        url += self.__gen_query_params_string(query_params)

        response = await request(
            url,
            protocol = 'HTTPS',
            protocol_info={
                "request_type": "GET",
                "headers": headers
            }
        )

        # TODO: Handle Exceptions

        result = response['api_response']

        if 200 <= result['status_code'] < 300 :
            return result['json']

    async def post(self, endpoint, path_params:dict = None, query_params:dict = None, json_payload:dict = {}):
        '''
        Post Request Endgine
        '''
        endpoint = endpoint
        
        # Format path parameter in endpoint url 
        if path_params:
            endpoint = endpoint.format(
                **path_params
            )
        
        url = urljoin(self.__base_url, endpoint)

        # Additional Header tokens added here
        headers = self.__headers()
        
        # Concatenate Qury Params
        url += self.__gen_query_params_string(query_params)

        response = await request(
            url,
            data=json_payload,
            protocol = 'HTTPS',
            protocol_info={
                "request_type": "POST",
                "headers": headers
            }
        )

        # TODO: Handle Exceptions
        result = response['api_response']

        if 200 <= result['status_code'] < 300 :
            return result['json']


    async def list_models(self, type:model.ModelType) -> List[model.Model]:
        '''
        '''
        results = await self.get(
           config.ENDPOINT_LIST['V2__MODEL_LIST'], 
            query_params={
                'type': type.value
            }
        )
        
        model_list = []
        for model_obj in results:
            model_list.append(
                model.Model(**model_obj)
            )
        return model_list
    
    def clear_model(self):
        '''
        '''
        self.__models = []

    def set_model(self, model:str):
        '''
        '''
        self.__models.append(model)
    
    async def check_job_status(self, job:model.Job) -> model.JobResponse:
        '''
        '''
        status = await self.get(
            config.ENDPOINT_LIST['V2__ASYNC_TEXT_STATUS'],
            path_params={
                'id': job.id
            }
        )
        return model.JobResponse(**status)

    async def text_gen(self, prompt:str, params:model.TextGenParams = model.TextGenParams()):
        '''
        '''
        request_data = model.TextGenRequest(
            models = self.__models,
            params = params,
            prompt = prompt
        )

        job_result = await self.post(
            config.ENDPOINT_LIST['V2__ASYNC_TEXT_SUBMIT'],
            json_payload = request_data.model_dump()
        )

        job = model.Job(
            **job_result
        )

        while True:
            job_status = await self.check_job_status(job)
            if job_status.done or not job_status.is_possible:
                return job_status
            await asyncio.sleep(config.REQUEST_RETRY_TIMEOUT)
horde_client/async_client.py
rahuldshetty/horde-client
import requests, random, time
from urllib.parse import  urljoin
from typing import List

from horde_client import config
from horde_client import model
from horde_client import errors

class HordeClient:
    def __init__(self, 
            base_url:str = config.API_BASE_URL,
            api_token:str = config.ANON_KEY,
            insecure:bool = False
        ):
        '''
        Instantiate HordeClient SDK.

        Parameters:
        -----------
        base_url (str):  KoboldAI API URL (Default: https://horde.koboldai.net/api/)
        api_token (str): KoboldAI API Key (Default: 0000000000)
        insecure (bool): Enabling insecure mode allows access to public KoboldAI. 
                         You will need to provide base_url when this flag is off (Default False)
        '''
        self.__base_url = base_url.strip()
        self.__api_token = api_token.strip()
        self.__insecure = insecure
        self.__models = []

        self.validate_insecure_permission()
    

    def validate_insecure_permission(self):
        '''
        To avoid accidental data leaks to public KoboldAI services, users have to manually enabled insecure mode.
        This methods is called during class initialization to validate the checks.
        '''
        if self.__insecure == False and self.__base_url == config.API_BASE_URL:
            raise errors.InsecureModePermissionError()
    
    def __headers(self):
        '''
        '''
        return {
            'Apikey': self.__api_token,
            'User-Agent': random.choice(config.user_agents)
        }

    def get(self, endpoint, path_params:dict = None, query_params:dict = None):
        '''
        Get Request Endgine
        '''
        endpoint = endpoint
        
        # Format path parameter in endpoint url 
        if path_params:
            endpoint = endpoint.format(
                **path_params
            )
        
        url = urljoin(self.__base_url, endpoint)

        # Additional Header tokens added here
        headers = self.__headers()

        response = requests.get(
            url,
            params = query_params,
            headers=headers
        )

        return response.json()

    def post(self, endpoint, path_params:dict = None, query_params:dict = None, json_payload:dict = {}):
        '''
        Post Request Endgine
        '''
        endpoint = endpoint
        
        # Format path parameter in endpoint url 
        if path_params:
            endpoint = endpoint.format(
                **path_params
            )
        
        url = urljoin(self.__base_url, endpoint)

        # Additional Header tokens added here
        headers = self.__headers()
        
        response = requests.post(
            url,
            params = query_params,
            headers=headers,
            json=json_payload
        )

        return response.json()


    def list_models(self, type:model.ModelType, sort_by:model.SortBy = model.SortBy.perf) -> List[model.Model]:
        '''
        '''
        results = self.get(
           config.ENDPOINT_LIST['V2__MODEL_LIST'], 
            query_params={
                'type': type.value
            }
        )
        model_list = []
        for model_obj in results:
            model_list.append(
                model.Model(**model_obj)
            )

        # Sort model results
        sort_by_fn = None
        reverse = False

        if sort_by == model.SortBy.perf:
            reverse = True
            sort_by_fn = lambda model: model.performance
        
        elif sort_by == model.SortBy.name:
            reverse = False
            sort_by_fn = lambda model: model.name

        model_list = sorted(model_list, key=sort_by_fn, reverse=reverse)

        return model_list
    
    def clear_model(self):
        '''
        '''
        self.__models = []

    def set_model(self, model:str):
        '''
        '''
        self.__models.append(model)
    
    def check_job_status(self, job:model.Job, type:model.ModelType, progress=False) -> model.JobResponse:
        '''
        Get latest status of Horde Async Job
        '''
        url = None

        if type == model.ModelType.text:
            url = config.ENDPOINT_LIST['V2__ASYNC_TEXT_STATUS']
        
        elif progress == True and type == model.ModelType.image:
            url = config.ENDPOINT_LIST['V2__ASYNC_IMAGE_PROGRESS_STATUS']
        elif progress == False and type == model.ModelType.image:
            url = config.ENDPOINT_LIST['V2__ASYNC_IMAGE_STATUS']

        status = self.get(
            url,
            path_params={
                'id': job.id
            }
        )
        return model.JobResponse(**status)

    def text_gen(self, prompt:str, params:model.TextGenParams = model.TextGenParams()) -> model.JobResponse:
        '''
        Leverage Horde LLM service to complete initial prompt using text generation.
        '''
        request_data = model.TextGenRequest(
            models = self.__models,
            params = params,
            prompt = prompt
        )

        job_result = self.post(
            config.ENDPOINT_LIST['V2__ASYNC_TEXT_SUBMIT'],
            json_payload = request_data.model_dump()
        )

        job = model.Job(
            **job_result
        )

        while True:
            job_status = self.check_job_status(job, model.ModelType.text)
            if job_status.done or not job_status.is_possible:
                return job_status
            time.sleep(config.REQUEST_RETRY_TIMEOUT)
        
    def image_gen(self, prompt: str, params:model.ImageGenParams = model.ImageGenParams()) -> model.JobResponse:
        '''
        Leverage Horde LLM service to generate image using an initial prompt.
        '''
        request_data = model.ImageGenRequest(
            models = self.__models,
            params = params,
            prompt = prompt
        )

        job_result = self.post(
            config.ENDPOINT_LIST['V2__ASYNC_IMAGE_SUBMIT'],
            json_payload = request_data.model_dump()
        )

        job = model.Job(
            **job_result
        )

        while True:
            job_status = self.check_job_status(job, model.ModelType.image, progress=True)

            if job_status.done or not job_status.is_possible:
                break
            time.sleep(config.REQUEST_RETRY_TIMEOUT)

        return self.check_job_status(job, model.ModelType.image)
        

horde_client/client.py
rahuldshetty/horde-client
API_BASE_URL = "https://horde.koboldai.net/api/"
ANON_KEY = '0000000000'
REQUEST_RETRY_TIMEOUT = 12

ENDPOINT_LIST = {
    "V2__MODEL_LIST": "v2/status/models",
    "V2__ASYNC_TEXT_SUBMIT": "v2/generate/text/async",
    "V2__ASYNC_TEXT_STATUS": "v2/generate/text/status/{id}",

    "V2__ASYNC_IMAGE_SUBMIT": "v2/generate/async",
    "V2__ASYNC_IMAGE_PROGRESS_STATUS": "v2/generate/check/{id}",
    "V2__ASYNC_IMAGE_STATUS": "v2/generate/status/{id}",
}

user_agents = [
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36'
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36'
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36'
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36'
    'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36'
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.1 Safari/605.1.15'
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 13_1) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.1 Safari/605.1.15'
]



horde_client/config.py
rahuldshetty/horde-client
from horde_client import config

class HordeClientException(Exception):
    pass

class InsecureModePermissionError(HordeClientException):
    def __init__(self, message="Due to security reasons, direct access to public KoboldAI services has been disabled. Enable `insecure` mode in the client settings to access the service."):
        super().__init__(message)

horde_client/errors.py
rahuldshetty/horde-client
from horde_client import config
from horde_client.model import TextGenParams
from horde_client.client import HordeClient
from horde_client.async_client import AsyncHordeClient


from typing import Any, List, Mapping, Optional

from langchain.callbacks.manager import CallbackManagerForLLMRun
from langchain.llms.base import LLM

class HordeClientLLM(LLM):
    base_url: str = config.API_BASE_URL
    """Base URL for KoboldAI Horde Service"""

    api_token: str = config.ANON_KEY
    """API Token for KoboldAI Horde Service"""
    
    insecure: bool = False
    """Flag controls access to public Horde service."""
    
    params: TextGenParams = TextGenParams()
    """Text Generation Parameter"""
    
    models: List[str] = []
    """Models to accept request."""

    @property
    def _llm_type(self) -> str:
        return "custom"

    def _call(
        self,
        prompt: str,
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> str:
        text = self.text_generate(prompt)

        if stop is not None:
            for sequence in stop:
                if text.endswith(sequence):
                    text = text[: -len(sequence)].rstrip()

        return text

    @property
    def _identifying_params(self) -> Mapping[str, Any]:
        """Get the identifying parameters."""
        return {
            "base_url": self.base_url,
            "api_token": self.api_token,
            "insecure": self.insecure,
            "params": self.params,
            "models": self.models
        }

    def text_generate(self, prompt):
        '''
        Synchronous Text Generation
        '''
        client = HordeClient(
            self.base_url,
            self.api_token,
            self.insecure
        )
        
        for model in self.models:
            client.set_model(model)
        
        return client.text_gen(
            prompt,
            self.params
        ).generations[0].text
horde_client/langchain_llm.py
rahuldshetty/horde-client
from enum import Enum
from pydantic import BaseModel
from typing import List

class SortBy(str, Enum):
    name = 'name'
    perf = 'perf'

class ModelType(str, Enum):
    text = 'text'
    image = 'image'

class Model(BaseModel):
    name: str
    count: int
    eta: int
    jobs: float
    performance: float
    queued: float
    type: ModelType

    def __str__(self) -> str:
        return "Model({name}, perf={perf})".format(
            name = self.name,
            perf = self.performance
        )

'''
Text Generation
'''

class TextGenParams(BaseModel):
    frmtadsnsp: bool = False
    frmtrmblln: bool = False
    frmtrmspch: bool = False
    frmttriminc: bool = False
    max_context_length: int = 1024
    max_length: int = 100
    n: int = 1
    rep_pen: float = 1.1
    rep_pen_range: int = 320
    rep_pen_slope: float = 0.7
    temperature: float = 0.7
    sampler_order: List[int | None] =  [6, 0, 1, 3, 4, 2, 5]
    tfs: int = 1
    top_a: int = 0
    top_k: int = 0
    top_p: float = 0.92
    typical: int = 1

class TextGenRequest(BaseModel):
    models: List[str]
    params: TextGenParams
    prompt: str
    use_default_badwordsids: bool = True
    workers: List[str] = []

class Job(BaseModel):
    id: str
    kudos: int

class JobGenerationOutputState(str, Enum):
    ok = "ok"
    censored = "censored"

class JobGenerationOutput(BaseModel):
    worker_id: str
    worker_name: str
    state: JobGenerationOutputState
    seed: int
    model: str

    # Text Generation Parameters
    text: str | None = None

    # Image Generation Parameters
    id: str | None = None
    img: str | None = None


class JobResponse(BaseModel):
    done: bool
    faulted: bool
    generations: List[JobGenerationOutput | None] = []
    is_possible: bool
    kudos: int
    processing: int
    queue_position: int
    restarted: int
    wait_time: int
    waiting: int

    def __str__(self) -> str:
        return "Output(done={done},responses=[{responses}], queue_position={queue_position})".format(
            done = str(self.done),
            responses = str(len(self.generations)),
            queue_position = str(self.queue_position)
        )


'''
Image Generation
'''

class ImageGenSampleType(str, Enum):
    k_lms = "k_lms"
    k_euler_a = "k_euler_a"

class ImageGenLora(BaseModel):
    name: str
    model: int = 1
    clip: int = 1
    inject_trigger: str = "string"

class ImageGenParams(BaseModel):
    width: int = 512
    height: int = 512
    
    n: int = 1
    steps: int = 20
    
    seed: str = ""
    sampler_name: str = ImageGenSampleType.k_euler_a
    karras: bool = False
    cfg_scale: float = 7.0
    post_processing: List[str | None] = []
    loras: List[ImageGenLora | None] = []


class ImageGenRequest(BaseModel):
    models: List[str]
    params: ImageGenParams
    prompt: str
    use_default_badwordsids: bool = True
    workers: List[str] = []

    censor_nsfw: bool = True
    nsfw: bool = False
    r2: bool = False
    replacement_filter: bool = True
horde_client/model.py
rahuldshetty/horde-client
from distutils.core import setup

# read the contents of your README file
from pathlib import Path
this_directory = Path(__file__).parent

long_description = (this_directory / "README.md").read_text()


setup(
    name='horde-client',
    version='1.0.3',
    description='Python Interface for AIHorde',
    author='Rahul D Shettu',
    author_email='35rahuldshetty@gmail.com',
    url='https://github.com/rahuldshetty/horde-client.git',
    install_requires=[
        'requests==2.27.1',
        'pydantic==2.3.0',
        'asyncio-requests==2.7.3',
        'langchain==0.0.291'
    ],
    packages=['horde_client'],
    long_description=long_description,
    long_description_content_type='text/markdown',
)
setup.py
rahuldshetty/horde-client