Modelos Personalizados de LangChain

Personalización de LLM

En el ámbito actual de modelos de IA, hay una gran variedad de modelos, y la integración oficial de LangChain no cubre todos los modelos. A veces puede que necesites personalizar un modelo e integrarlo en el marco de LangChain.

Este capítulo presentará cómo crear un envoltorio personalizado para LLM, haciéndolo conveniente para que uses tu propio modelo o modelos no compatibles con LangChain.

En LangChain, si deseas usar tu propio LLM o un envoltorio diferente al admitido por LangChain, puedes crear un envoltorio personalizado para LLM. Un LLM personalizado solo necesita implementar dos métodos requeridos:

• Un método _call, que toma una cadena como entrada, algunas palabras de detención opcionales y devuelve una cadena, implementando la invocación del modelo en el método _call.
• Un atributo _llm_type, que devuelve una cadena que representa el nombre del modelo, utilizado únicamente con fines de registro.

Además de los métodos requeridos, un LLM personalizado también puede implementar un método opcional:

• Un atributo _identifying_params, utilizado para ayudar a imprimir la clase. Debería devolver un diccionario.

Implementación de un LLM personalizado simple

Vamos a implementar un LLM personalizado muy simple que solo devuelve los primeros n caracteres de la entrada.

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("No se permiten palabras de detención.")
        return prompt[: self.n]

    @property
    def _identifying_params(self) -> Mapping[str, Any]:
        """Obtener los parámetros de identificación."""
        return {"n": self.n}

Ahora podemos usar este LLM personalizado como cualquier otro LLM.

Uso del LLM personalizado

Podemos instanciar y usar el objeto LLM personalizado, demostrando cómo invocar el LLM personalizado y personalizar la salida impresa.

llm = CustomLLM(n=10)

llm.invoke("Esto es algo genial")

'Esto es algo'

También podemos imprimir el LLM y ver su salida personalizada.

print(llm)

CustomLLM
Parámetros: {'n': 10}

Personalización del modelo de chat

Aquí explicaremos cómo personalizar el modelo de chat de LangChain.

Entrada y salida de mensajes

En el modelo de chat, los mensajes son el foco de entrada y salida. Un mensaje es el contenido ingresado por el usuario y la respuesta generada por el modelo.

Mensajes

El modelo de chat toma mensajes como entrada y luego genera uno o más mensajes como salida. En LangChain, hay varios tipos de mensajes integrados, incluyendo:

• SystemMessage: utilizado para inicializar el comportamiento de la IA, generalmente como el primer mensaje en una serie de mensajes de entrada.
• HumanMessage: representa la interacción del usuario con el modelo de chat.
• AIMessage: representa mensajes del modelo de chat, que pueden ser texto o solicitudes de invocación de herramientas.
• FunctionMessage / ToolMessage: utilizado para devolver el resultado de una invocación de herramientas al modelo.

El uso de estos tipos de mensajes se puede ampliar y personalizar según requisitos específicos, como ajustar según los parámetros function y tool de OpenAI.

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

Variantes fluidas

Todos los mensajes de chat tienen una variante fluida con Chunk en sus nombres.

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

Estos Chunk se utilizan al transmitir el modelo de chat y cada uno de ellos define un atributo acumulativo.

Ejemplo

AIMessageChunk(content="Hola") + AIMessageChunk(content=" mundo!")

Devuelve

AIMessageChunk(content='¡Hola mundo!')

Modelo de Chat Simple

Heredar de SimpleChatModel permite implementar rápidamente un modelo de chat simple.

Aunque puede que no incorpore todas las funcionalidades necesarias para un modelo de chat, proporciona una implementación rápida. Si se requieren más características, es posible hacer la transición al BaseChatModel descrito a continuación.

Heredar de SimpleChatModel requiere la implementación de la siguiente interfaz:

• Método _call - implementación de llamadas a la API del modelo externo.

Además, se pueden especificar lo siguiente:

• Atributo _identifying_params - utilizado para registrar información parametrizada del modelo.

Opcionalmente:

• Método _stream - utilizado para implementar la salida en tiempo real.

Modelo Base de Chat

Heredar de BaseChatModel requiere la implementación del método _generate y el atributo _llm_type. Opcionalmente, es posible la implementación de _stream, _agenerate, _astream y _identifying_params.

Ejemplo de un Modelo de Chat Personalizado

En esta sección, demostraremos la implementación de código de un modelo de chat personalizado llamado CustomChatModelAdvanced, que incluye la generación de resultados de chat, la salida en tiempo real y la implementación de flujo asincrónico.

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 modelo de chat personalizado que devuelve los primeros `n` caracteres del último mensaje."""


    n: int
    """Parámetro del modelo personalizado"""

    def _generate(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> ChatResult:
        """Aquí se implementa la lógica de llamada al modelo, generalmente llamando a la API de un modelo de terceros, y luego encapsulando el resultado devuelto por la API en un formato que pueda reconocer langchain.
        Explicación clave de los parámetros:
            messages: Una lista de indicaciones compuestas por mensajes
        """
        ultimo_mensaje = messages[-1]
        tokens = ultimo_mensaje.content[: self.n]
        mensaje = AIMessage(content=tokens)
        generación = ChatGeneration(message=mensaje)
        return ChatResult(generations=[generación])

    def _stream(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> Iterator[ChatGenerationChunk]:
        """Implementación de la salida en tiempo real del modelo, similar al método _generate, pero con procesamiento de salida en tiempo real."""
        ultimo_mensaje = messages[-1]
        tokens = ultimo_mensaje.content[: self.n]

        for token in tokens:
            chunk = ChatGenerationChunk(message=AIMessageChunk(content=token))

            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]:
        """Versión asincrónica de la implementación del método `stream`"""
        resultado = 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 resultado:
            yield chunk

    @property
    def _llm_type(self) -> str:
        """Devuelve la etiqueta del modelo personalizado"""
        return "modelo-de-chat-avanzado-de-eco"

    @property
    def _identifying_params(self) -> Dict[str, Any]:
        """Devuelve información de depuración personalizada"""
        return {"n": self.n}

Probando el Modelo de Chat Personalizado

Probemos el modelo de chat, incluyendo el uso de los métodos invoke, batch, stream y la implementación de flujo asincrónico.

modelo = CustomChatModelAdvanced(n=3)

modelo.invoke([HumanMessage(content="¡Hola!")])

modelo.invoke("hola")

modelo.batch(["hola", "adiós"])

for chunk in modelo.stream("gato"):
    print(chunk.content, end="|")

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

async for evento in modelo.astream_events("gato", version="v1"):
    print(evento)