Personnalisation de LLM

Dans le domaine actuel des modèles d'IA, il existe une grande variété de modèles, et l'intégration officielle de LangChain ne couvre pas tous les modèles. Parfois, vous devrez peut-être personnaliser un modèle et l'intégrer dans le cadre de LangChain.

Ce chapitre présentera comment créer un wrapper LLM personnalisé, ce qui vous permettra d'utiliser votre propre modèle ou des modèles non pris en charge par LangChain.

Dans LangChain, si vous souhaitez utiliser votre propre LLM ou un wrapper différent de celui pris en charge par LangChain, vous pouvez créer un wrapper LLM personnalisé. Un LLM personnalisé n'a besoin que de mettre en œuvre deux méthodes requises :

  • Une méthode _call, qui prend une chaîne en entrée, éventuellement des mots d'arrêt optionnels, et renvoie une chaîne, mettant en œuvre l'invocation du modèle dans la méthode _call.
  • Un attribut _llm_type, renvoyant une chaîne représentant le nom du modèle, utilisé uniquement à des fins de journalisation.

En plus des méthodes requises, un LLM personnalisé peut également mettre en œuvre une méthode facultative :

  • Un attribut _identifying_params, utilisé pour aider à imprimer la classe. Il devrait renvoyer un dictionnaire.

Mise en œuvre d'un LLM personnalisé simple

Implémentons un LLM personnalisé très simple qui ne renvoie que les premiers caractères de l'entrée.

from typing import Any, List, Mapping, Optional

from langchain_core.callbacks.manager import CallbackManagerForLLMRun
from langchain_core.language_models.llms import LLM

class CustomLLM(LLM):
    n: int

    @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:
        if stop is not None:
            raise ValueError("les arguments d'arrêt ne sont pas autorisés.")
        return prompt[: self.n]

    @property
    def _identifying_params(self) -> Mapping[str, Any]:
        """Obtenir les paramètres d'identification."""
        return {"n": self.n}

Maintenant, nous pouvons utiliser ce LLM personnalisé comme n'importe quel autre LLM.

Utilisation du LLM personnalisé

Nous pouvons instancier et utiliser l'objet LLM personnalisé, démontrant comment invoquer le LLM personnalisé et personnaliser la sortie de l'impression.

llm = CustomLLM(n=10)

llm.invoke("C'est une chose foobar")

'C'est une c'

Nous pouvons également imprimer le LLM et voir sa sortie d'impression personnalisée.

print(llm)

CustomLLM
Paramètres: {'n': 10}

Personnalisation du modèle de chat

Ici, nous expliquerons comment personnaliser le modèle de chat de LangChain.

Entrée et sortie des messages

Dans le modèle de chat, les messages sont au centre de l'entrée et de la sortie. Un message est le contenu saisi par l'utilisateur et la réponse générée par le modèle.

Messages

Le modèle de chat prend des messages en entrée, puis génère un ou plusieurs messages en sortie. Dans LangChain, il existe plusieurs types de messages intégrés, notamment :

  • SystemMessage : Utilisé pour initialiser le comportement de l'IA, généralement comme premier message dans une série de messages d'entrée.
  • HumanMessage : Représente l'interaction de l'utilisateur avec le modèle de chat.
  • AIMessage : Représente les messages du modèle de chat, qui peuvent être du texte ou des demandes d'invocation d'outils.
  • FunctionMessage / ToolMessage : Utilisé pour renvoyer le résultat d'une invocation d'outil au modèle.

L'utilisation de ces types de messages peut être étendue et personnalisée en fonction des besoins spécifiques, tels que l'ajustement selon les paramètres function et tool d'OpenAI.

from langchain_core.messages import (
    AIMessage,
    BaseMessage,
    FunctionMessage,
    HumanMessage,
    SystemMessage,
)

Variantes fluides

Tous les messages de chat possèdent une variante fluide avec Chunk dans leur nom.

from langchain_core.messages import (
    AIMessageChunk,
    FunctionMessageChunk,
    HumanMessageChunk,
    SystemMessageChunk,
    ToolMessageChunk,
)

Ces Chunk sont utilisés lors de la diffusion du modèle de chat et chacun définit un attribut cumulatif !

Exemple

AIMessageChunk(content="Bonjour") + AIMessageChunk(content=" tout le monde !")

Renvoie

AIMessageChunk(content='Bonjour tout le monde!')

Modèle de chat simple

L'héritage de SimpleChatModel permet d'implémenter rapidement un modèle de chat simple.

Bien qu'il ne contienne pas toutes les fonctionnalités nécessaires pour un modèle de chat, il permet une mise en œuvre rapide. Si davantage de fonctionnalités sont nécessaires, il est possible de passer au BaseChatModel décrit ci-dessous.

L'héritage de SimpleChatModel nécessite l'implémentation de l'interface suivante :

  • Méthode _call - implémentation des appels d'API de modèle externe.

De plus, les éléments suivants peuvent être spécifiés :

  • Attribut _identifying_params - utilisé pour enregistrer les informations paramétrées du modèle.

Facultatif :

  • Méthode _stream - utilisée pour implémenter une sortie en continu.

Modèle de chat de base

L'héritage de BaseChatModel nécessite l'implémentation de la méthode _generate et de l'attribut _llm_type. Facultativement, l'implémentation de _stream, _agenerate, _astream et _identifying_params est possible.

Exemple d'un modèle de chat personnalisé

Dans cette section, nous allons présenter l'implémentation de code d'un modèle de chat personnalisé appelé CustomChatModelAdvanced, comprenant la génération de résultats de chat, la sortie en continu et l'implémentation de la sortie en continu asynchrone.

from typing import Any, AsyncIterator, Dict, Iterator, List, Optional

from langchain_core.callbacks import (
    AsyncCallbackManagerForLLMRun,
    CallbackManagerForLLMRun,
)
from langchain_core.language_models import BaseChatModel, SimpleChatModel
from langchain_core.messages import AIMessageChunk, BaseMessage, HumanMessage
from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, ChatResult
from langchain_core.runnables import run_in_executor


class CustomChatModelAdvanced(BaseChatModel):
    """Implémenter un modèle de chat personnalisé qui renvoie les premiers `n` caractères du dernier message."""

    n: int
    """Paramètre de modèle personnalisé"""

    def _generate(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> ChatResult:
        """Ici, la logique d'appel du modèle est implémentée, généralement en appelant l'API d'un modèle tiers, puis en encapsulant le résultat retourné par l'API dans un format que langchain peut reconnaître.
        Explication des paramètres clés :
            messages : Une liste de prompts composés de messages
        """
        dernier_message = messages[-1]
        jetons = dernier_message.content[: self.n]
        message = AIMessage(content=jetons)
        generation = ChatGeneration(message=message)
        return ChatResult(generations=[generation])

    def _stream(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> Iterator[ChatGenerationChunk]:
        """Implémentation de la sortie en continu du modèle, similaire à la méthode _generate, mais avec le traitement de la sortie en continu."""
        dernier_message = messages[-1]
        jetons = dernier_message.content[: self.n]

        for jeton in jetons:
            chunk = ChatGenerationChunk(message=AIMessageChunk(content=jeton))

            if run_manager:
                run_manager.on_llm_new_token(token, chunk=chunk)

            yield chunk

    async def _astream(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[AsyncCallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> AsyncIterator[ChatGenerationChunk]:
        """Version asynchrone de l'implémentation de la méthode `stream`"""
        resultat = await run_in_executor(
            None,
            self._stream,
            messages,
            stop=stop,
            run_manager=run_manager.get_sync() if run_manager else None,
            **kwargs,
        )
        for chunk in resultat:
            yield chunk

    @property
    def _llm_type(self) -> str:
        """Renvoyer le tag du modèle personnalisé"""
        return "echoing-chat-model-advanced"

    @property
    def _identifying_params(self) -> Dict[str, Any]:
        """Renvoyer des informations de débogage personnalisées"""
        return {"n": self.n}

Test du modèle de chat personnalisé

Testons le modèle de chat, y compris l'utilisation des méthodes invoke, batch, stream et l'implémentation de la sortie en continu asynchrone.

model = CustomChatModelAdvanced(n=3)

model.invoke([HumanMessage(content="hello!")])

model.invoke("hello")

model.batch(["hello", "goodbye"])

for chunk in model.stream("cat"):
    print(chunk.content, end="|")

async for chunk in model.astream("cat"):
    print(chunk.content, end="|")

async for event in model.astream_events("chat", version="v1"):
    print(event)