DjangoによるOpenGPTsの実現【構成検討編】

悩んでいる人

LLM（大規模言語モデル）を用いたアプリケーションとしてOpenGPTsがあるが、1サーバ1クライアントになっているため使いづらい。

1サーバで複数のクライアントが利用できるような構成を教えて欲しい。

今回は、上記のお悩みに対し、複数クライアントを構成する方法について検討します。

OpenGPTsの詳細はこちらのgithubを参照してください。

構成案

モデル構成の略図は以下に示す通りです。また、モデルが持つ各機能（メソッド）は省略しております。

それぞれのモデルの詳細は以下に示す通りです。

対象モデル	概要	備考
User	Djangoで扱えるユーザモデル他のライブラリと共存できるように、profileフィールドを持つ。	profileフィールドはOneToOneFieldとして定義される。
Profile	各社が提供するLLMモデルのAPI情報（APIキーやエンドポイントなど）を管理するモデル	現状、OpenAI、AzureOpenAIなど対象の言語モデルごとにフィールドを定義する必要あり
ModelType	モデルの種別	DjangoのIntegerChoiceFieldにより実装される想定
OpenAI	OpenAI用のAPI情報	各Modelに応じたLLMインスタンスを返却するメソッドを持つ。
AzureOpenAI	AzureOpenAI用のAPI情報	各Modelに応じたLLMインスタンスを返却するメソッドを持つ。

それぞれのモデルの詳細

構成案を概要レベルで定義できた段階で、次はDjangoによるモデル定義例を示します。

Djangoによるモデル定義例

以降では、モデルごとの実装例を示します。

最初に、想定するディレクトリ構成を以下に示します。（モデル以外は省略します）


backend/
|-- account/
|   `-- models.py
`-- opengpts/
    `-- models/
        |-- profile.py
        `-- llms.py

ユーザモデル（User）

account/models.pyで定義するユーザモデル（User）の構成を以下に示します。

from django.db import models
from django.core.mail import send_mail
from django.contrib.auth.models import PermissionsMixin, UserManager
from django.contrib.auth.base_user import AbstractBaseUser
from django.utils import timezone
from django.contrib.auth.validators import UnicodeUsernameValidator
from django.utils.translation import gettext_lazy

def get_display_name():
    now = timezone.now()
    _localtime = timezone.localtime(now)
    output = _localtime.strftime('user%Y%m%d%H%M%S%f')

    return output

class CustomUserManager(UserManager):
    use_in_migrations = True

    def _create_user(self, username, email, password, **extra_fields):
        if not username:
            raise ValueError('The given username must be set')
        if not email:
            raise ValueError('The given email must be set')
        username = self.model.normalize_username(username)
        email = self.normalize_email(email)
        user = self.model(username=username, email=email, **extra_fields)
        user.set_password(password)
        user.save(using=self._db)

        return user

    def create_user(self, username, email, password=None, **extra_fields):
        extra_fields.setdefault('is_staff', False)
        extra_fields.setdefault('is_superuser', False)

        return self._create_user(username, email, password,  **extra_fields)

    def create_superuser(self, username, email, password=None, **extra_fields):
        extra_fields.setdefault('is_staff', True)
        extra_fields.setdefault('is_superuser', True)

        if extra_fields.get('is_staff') is not True:
            raise ValueError('Superuser must have is_staff=True.')
        if extra_fields.get('is_superuser') is not True:
            raise ValueError('Superuser must have is_superuser=True.')

        return self._create_user(username, email, password,  **extra_fields)

class User(AbstractBaseUser, PermissionsMixin):
    username_validator = UnicodeUsernameValidator()

    username = models.CharField(
        gettext_lazy('username'),
        max_length=255,
        unique=True,
        help_text=gettext_lazy('Required. 255 characters allowing only Unicode characters, in addition to @, ., +, -, and _.'),
        validators=[username_validator],
        error_messages={
            'unique': gettext_lazy("A user with that username already exists."),
        },
    )
    display_name = models.CharField(
        gettext_lazy('display name'),
        max_length=255,
        blank=False,
        default=get_display_name,
        help_text=gettext_lazy('Required. 255 characters or fewer.'),
    )
    email = models.EmailField(gettext_lazy('email address'), unique=True)
    is_staff = models.BooleanField(
        gettext_lazy('staff status'),
        default=False,
        help_text=gettext_lazy('Designates whether the user can log into this admin site.'),
    )
    is_active = models.BooleanField(
        gettext_lazy('active'),
        default=True,
        help_text=gettext_lazy(
            'Designates whether this user should be treated as active. Unselect this instead of deleting accounts.'
        ),
    )
    date_joined = models.DateTimeField(gettext_lazy('date joined'), default=timezone.now)

    objects = CustomUserManager()

    EMAIL_FIELD = 'email'
    USERNAME_FIELD = 'username'
    REQUIRED_FIELDS = ['email']

    class Meta:
        verbose_name = gettext_lazy('user')
        verbose_name_plural = gettext_lazy('users')

    def __str__(self):
        return self.display_name

    def __unicode__(self):
        return str(self)

    def get_full_name(self):
        return str(self)

    def get_short_name(self):
        max_len = 16
        fullname = self.get_full_name()
        shortname = '{}...'.format(fullname[:max_len]) if len(fullname) > max_len else fullname

        return shortname

    def email_user(self, subject, message, from_email=None, **kwargs):
        send_mail(subject, message, from_email, [self.email], **kwargs)

Djangoで良く使われる形式のユーザモデルとなります。

LLMsモデル

次に、各LLMモデルを定義します。

ModelType、OpenAI、AzureOpenAIはそれぞれ、以下のようになります。

import logging
import httpx
from urllib.parse import urlparse
# for LangChain
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_openai import AzureChatOpenAI, AzureOpenAIEmbeddings
# for Django
from django.db import models
from django.utils.translation import gettext_lazy
from django.contrib.auth import get_user_model

User = get_user_model()
logger = logging.getLogger(__name__)

def _get_client(proxy_url: str = None, is_async: bool = False):
  http_client = None

  if proxy_url is not None and proxy_url:
    parsed_url = urlparse(proxy_url)

    if parsed_url.scheme and parsed_url.netloc:
      if is_async:
        http_client = httpx.AsyncClient(proxies=proxy_url)
      else:
        http_client = httpx.Client(proxies=proxy_url)
    else:
      logger.warn('Invalid proxy URL provided. Proceeding without proxy.')

  return http_client

class ModelType(models.IntegerChoices):
  # [format] name = value, label
  OpenAI          = 1, gettext_lazy('OpenAI')
  AzureOpenAI     = 2, gettext_lazy('Azure OpenAI')
  EmbeddingOpenAI = 3, gettext_lazy('Embedding of OpenAI')
  EmbeddingAzure  = 3, gettext_lazy('Embedding of Azure OpenAI')

  @classmethod
  def get_openai_choices(cls):
    fields = [cls.OpenAI, cls.EmbeddingOpenAI]
    choices = [(field.value, field.label) for field in fields]

    return choices
  
  @classmethod
  def get_azure_choices(cls):
    fields = [cls.AzureOpenAI, cls.EmbeddingAzure]
    choices = [(field.value, field.label) for field in fields]

    return choices

class OpenAI(models.Model):
  name = models.CharField(max_length=128)
  type = models.IntegerField(
    gettext_lazy('Model Type'),
    choices=ModelType.get_openai_choices(),
    default=ModelType.OpenAI,
  )
  api_key = models.CharField(max_length=255)
  endpoint = models.CharField(max_length=255)
  api_version = models.CharField(max_length=255)
  model_name = models.CharField(max_length=255)
  deployment_name = models.CharField(max_length=255)
  user = models.ForeignKey(
    User,
    on_delete=models.CASCADE,
    verbose_name=gettext_lazy('User'),
    related_name='openais',
  )

  def __str__(self):
    return str(ModelType(self.type).label)

  def get_chatbot(self, proxy_url: str = None):
    llm = ChatOpenAI(
      http_client=_get_client(proxy_url, is_async=False),
      http_async_client=_get_client(proxy_url, is_async=True),
      model=self.model_name,
      base_url=self.endpoint,
      api_key=self.api_key,
      max_retries=10,
      temperature=0,
    )

    return llm

  def get_embedding(self, proxy_url: str = None):
    embedding = OpenAIEmbeddings(
      http_client=_get_client(proxy_url, is_async=False),
      http_async_client=_get_client(proxy_url, is_async=True),
      deployment=self.deployment_name,
      model=self.model_name,
      base_url=self.endpoint,
      api_key=self.api_key,
      api_version=self.api_version,
    )

    return embedding

class AzureOpenAI(models.Model):
  name = models.CharField(max_length=128)
  type = models.IntegerField(
    gettext_lazy('Model Type'),
    choices=ModelType.get_azure_choices(),
    default=ModelType.AzureOpenAI,
  )
  api_key = models.CharField(max_length=255)
  endpoint = models.CharField(max_length=255)
  api_version = models.CharField(max_length=255)
  model_name = models.CharField(max_length=255)
  deployment_name = models.CharField(max_length=255)
  user = models.ForeignKey(
    User,
    on_delete=models.CASCADE,
    verbose_name=gettext_lazy('User'),
    related_name='azures',
  )

  def get_chatbot(self, proxy_url: str = None):
    llm = AzureChatOpenAI(
      openai_api_type='azure',
      http_client=_get_client(proxy_url, is_async=False),
      http_async_client=_get_client(proxy_url, is_async=True),
      deployment_name=self.deployment_name,
      model=self.model_name,
      azure_endpoint=self.endpoint,
      api_key=self.api_key,
      api_version=self.api_version,
      max_retries=10,
      temperature=0,
    )

    return llm

  def get_embedding(self, proxy_url: str = None):
    embedding = AzureOpenAIEmbeddings(
      openai_api_type='azure',
      http_client=_get_client(proxy_url, is_async=False),
      http_async_client=_get_client(proxy_url, is_async=True),
      deployment=self.deployment_name,
      model=self.model_name,
      azure_endpoint=self.endpoint,
      api_key=self.api_key,
      api_version=self.api_version,
    )

    return embedding

Profileモデル

最後に、Profileモデルを定義します。

from django.db import models
from django.utils.translation import gettext_lazy
from django.contrib.auth import get_user_model
from .opengpts.models import OpenAI, AzureOpenAI

User = get_user_model()

class Profile(models.Model):
  user = models.OneToOneField(User)
  openai = models.ManyToManyField(
    OpenAI,
    on_delete=models.CASCADE,
    verbose_name=gettext_lazy('OpenAI'),
    related_name='openai_users',
  )
  azure = models.ManyToManyField(
    AzureOpenAI,
    on_delete=models.CASCADE,
    verbose_name=gettext_lazy('Azure OpenAI'),
    related_name='azure_users',
  )

今後の検討事項

今回は、1サーバ複数クライアントを実現するための構成案を検討しました。

実際にOpenGPTsとして動作させるためには、Agentの定義、チャット形式でメッセージを受信した際のstreamの扱いなど、検討すべきことが多く残っています。

詳細を検討し次第、アップデート予定です。また、ある程度構成が出来上がった段階で、実際に実装してみたいと思います。