Modelli personalizzati LangChain

Personalizzazione di LLM

Nel attuale dominio dei modelli di intelligenza artificiale, esiste una grande varietà di modelli e l'integrazione ufficiale di LangChain non copre tutti i modelli. A volte potresti aver bisogno di personalizzare un modello e integrarlo nel framework di LangChain.

Questo capitolo introdurrà come creare un wrapper LLM personalizzato, rendendolo comodo per te utilizzare il tuo modello o modelli non supportati da LangChain.

In LangChain, se desideri utilizzare il tuo LLM personale o un wrapper diverso da quello supportato da LangChain, puoi creare un wrapper LLM personalizzato. Un LLM personalizzato deve solo implementare due metodi richiesti:

• Un metodo _call, che prende una stringa in input, alcune parole stop opzionali e restituisce una stringa, implementando l'invocazione del modello nel metodo _call.
• Un attributo _llm_type, che restituisce una stringa rappresentante il nome del modello, utilizzato esclusivamente per scopi di logging.

Oltre ai metodi richiesti, un LLM personalizzato può anche implementare un metodo opzionale:

• Un attributo _identifying_params, utilizzato per aiutare a stampare la classe. Dovrebbe restituire un dizionario.

Implementazione di un semplice LLM personalizzato

Implementiamo un LLM personalizzato molto semplice che restituisce solo i primi caratteri dell'input.

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 "personalizzato"

    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("I parametri stop non sono consentiti.")
        return prompt[: self.n]

    @property
    def _identifying_params(self) -> Mapping[str, Any]:
        """Ottieni i parametri identificativi."""
        return {"n": self.n}

Ora possiamo utilizzare questo LLM personalizzato esattamente come qualsiasi altro LLM.

Utilizzo del LLM personalizzato

Possiamo istanziare e utilizzare l'oggetto LLM personalizzato, dimostrando come invocare il LLM personalizzato e personalizzare l'output della stampa.

llm = CustomLLM(n=10)

llm.invoke("Questa è una cosa fantastica")

'Questa è u'

Possiamo anche stampare l'LLM e visualizzare la sua stampa personalizzata.

print(llm)

CustomLLM
Parametri: {'n': 10}

Personalizzazione del Modello di Chat

Qui spiegheremo come personalizzare il modello di chat di LangChain.

Input e Output del Messaggio

Nel modello di chat, i messaggi sono il focus dell'input e dell'output. Un messaggio è il contenuto inserito dall'utente e la risposta generata dal modello.

Messaggi

Il modello di chat prende messaggi in input e poi genera uno o più messaggi in output. In LangChain, ci sono diversi tipi di messaggi incorporati, tra cui:

• SystemMessage: Utilizzato per inizializzare il comportamento dell'IA, di solito come primo messaggio in una serie di messaggi di input.
• HumanMessage: Rappresenta l'interazione dell'utente con il modello di chat.
• AIMessage: Rappresenta i messaggi dal modello di chat, che possono essere testo o richieste di invocazione di strumenti.
• FunctionMessage / ToolMessage: Utilizzato per passare il risultato di un'invocazione di strumenti al modello.

L'uso di questi tipi di messaggi può essere espanso e personalizzato in base a requisiti specifici, come adattamento basato sui parametri function e tool di OpenAI.

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

Varianti Fluenti

Tutti i messaggi di chat hanno una variante fluente con Chunk nei loro nomi.

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

Questi Chunk vengono utilizzati durante lo streaming del modello di chat e ognuno di essi definisce un attributo cumulativo!

Esempio

AIMessageChunk(content="Ciao") + AIMessageChunk(content=" mondo!")

Restituisce

AIMessageChunk(content='Ciao mondo!')

Modello di Chat Semplice

L'eredità da SimpleChatModel consente di implementare rapidamente un modello di chat semplice.

Anche se potrebbe non incorporare tutte le funzionalità necessarie per un modello di chat, fornisce un'implementazione rapida. Se sono richieste più funzionalità, è possibile passare al BaseChatModel descritto di seguito.

L'eredità da SimpleChatModel richiede l'implementazione dell'interfaccia seguente:

• Metodo _call - implementazione delle chiamate API del modello esterno.

Inoltre, può essere specificato quanto segue:

• Attributo _identifying_params - utilizzato per registrare le informazioni parametrate del modello.

Opzionale:

• Metodo _stream - utilizzato per implementare l'output in streaming.

Base Modello di Chat

L'eredità da BaseChatModel richiede l'implementazione del metodo _generate e dell'attributo _llm_type. Opzionalmente, è possibile implementare _stream, _agenerate, _astream e _identifying_params.

Esempio di un Modello di Chat Personalizzato

In questa sezione, mostreremo l'implementazione del codice di un modello di chat personalizzato chiamato CustomChatModelAdvanced, inclusa la generazione di risultati della chat, l'output in streaming e l'implementazione dello streaming asincrono.

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):
    """Implementa un modello di chat personalizzato che restituisce i primi `n` caratteri dell'ultimo messaggio."""


    n: int
    """Parametro del modello personalizzato"""

    def _generate(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> ChatResult:
        """Qui è implementata la logica della chiamata al modello, di solito chiamando l'API di un modello di terze parti e quindi incapsulando il risultato restituito dall'API in un formato che langchain può riconoscere.
        Spiegazione chiave del parametro:
            messages: Una lista di prompt composti da messaggi
        """
        ultimo_messaggio = messages[-1]
        token = ultimo_messaggio.content[: self.n]
        messaggio = AIMessage(content=token)
        generazione = ChatGeneration(messaggio=messaggio)
        return ChatResult(generazioni=[generazione])

    def _stream(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> Iterator[ChatGenerationChunk]:
        """Implementazione dell'output in streaming del modello, simile al metodo _generate, ma con elaborazione dell'output in streaming."""
        ultimo_messaggio = messages[-1]
        token = ultimo_messaggio.content[: self.n]

        for token in token:
            chunk = ChatGenerationChunk(messaggio=AIMessageChunk(content=token))

            if run_manager:
                run_manager.su_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]:
        """Versione asincrona dell'implementazione del metodo `stream`"""
        risultato = attendi run_in_executor(
            None,
            self._stream,
            messages,
            stop=stop,
            run_manager=run_manager.get_sync() if run_manager else None,
            **kwargs,
        )
        per chunk in risultato:
            yield chunk

    @property
    def _llm_type(self) -> str:
        """Restituisci il tag del modello personalizzato"""
        return "modello-chat-avanzato-di-eco"

    @property
    def _identifying_params(self) -> Dict[str, Any]:
        """Restituisci informazioni di debug personalizzate"""
        return {"n": self.n}

Test del Modello di Chat Personalizzato

Testiamo il modello di chat, inclusi l'utilizzo dei metodi invoke, batch, stream e l'implementazione dello streaming asincrono.

modello = CustomChatModelAdvanced(n=3)

modello.invoke([HumanMessage(content="ciao!")])

modello.invoke("ciao")

modello.batch(["ciao", "arrivederci"])

per chunk in modello.stream("gatto"):
    stampa(chunk.content, end="|")

async for chunk in modello.astream("gatto"):
    stampa(chunk.content, end="|")

async for evento in modello.astream_events("gatto", versione="v1"):
    stampa(evento)