Modelos Personalizados do LangChain

Personalizando LLM

No domínio atual dos modelos de IA, existe uma grande variedade de modelos e a integração oficial da LangChain não abrange todos os modelos. Às vezes, pode ser necessário personalizar um modelo e integrá-lo ao framework LangChain.

Este capítulo apresentará como criar um envoltório LLM personalizado, tornando conveniente o uso do seu próprio modelo ou de modelos não suportados pela LangChain.

Na LangChain, se desejar usar seu próprio LLM ou um envoltório diferente daquele suportado pela LangChain, você pode criar um envoltório LLM personalizado. Um LLM personalizado só precisa implementar dois métodos necessários:

• Um método _call, que recebe uma string como entrada, algumas palavras de parada opcionais e retorna uma string, implementando a invocação do modelo no método _call.
• Um atributo _llm_type, que retorna uma string representando o nome do modelo, usado apenas para fins de registro.

Além dos métodos necessários, um LLM personalizado também pode implementar um método opcional:

• Um atributo _identifying_params, usado para ajudar na impressão da classe. Deve retornar um dicionário.

Implementando um LLM personalizado simples

Vamos implementar um LLM personalizado muito simples que retorna apenas os primeiros n caracteres da 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("stop kwargs are not permitted.")
        return prompt[: self.n]

    @property
    def _identifying_params(self) -> Mapping[str, Any]:
        """Obter os parâmetros de identificação."""
        return {"n": self.n}

Agora podemos usar este LLM personalizado como qualquer outro LLM.

Usando o LLM personalizado

Podemos instanciar e usar o objeto LLM personalizado, demonstrando como invocar o LLM personalizado e personalizar a saída da impressão.

llm = CustomLLM(n=10)

llm.invoke("This is a foobar thing")

'This is a '

Também podemos imprimir o LLM e visualizar sua saída de impressão personalizada.

print(llm)

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

Personalizando o Modelo de Chat

Aqui explicaremos como personalizar o modelo de chat da LangChain.

Entrada e Saída de Mensagens

No modelo de chat, as mensagens são o foco de entrada e saída. Uma mensagem é o conteúdo inserido pelo usuário e a resposta gerada pelo modelo.

Mensagens

O modelo de chat recebe mensagens como entrada e em seguida gera uma ou mais mensagens como saída. Na LangChain, existem vários tipos de mensagens incorporados, incluindo:

• SystemMessage: Usado para inicializar o comportamento da IA, geralmente como a primeira mensagem em uma série de mensagens de entrada.
• HumanMessage: Representa a interação do usuário com o modelo de chat.
• AIMessage: Representa mensagens do modelo de chat, que podem ser texto ou solicitações de invocação de ferramenta.
• FunctionMessage / ToolMessage: Usado para passar o resultado de uma invocação de ferramenta de volta ao modelo.

O uso desses tipos de mensagens pode ser expandido e personalizado de acordo com requisitos específicos, como ajustar com base nos parâmetros function e tool da OpenAI.

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

Variantes Fluentes

Todas as mensagens de chat têm uma variante fluente com Chunk em seus nomes.

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

Estes Chunk são usados ao transmitir o modelo de chat e cada um deles define um atributo cumulativo!

Exemplo

AIMessageChunk(content="Olá") + AIMessageChunk(content=" mundo!")

Retorna

AIMessageChunk(content='Olá mundo!')

Modelo de Chat Simples

A herança de SimpleChatModel permite implementar rapidamente um modelo de chat simples.

Embora não incorpore todas as funcionalidades necessárias para um modelo de chat, proporciona uma implementação ágil. Se mais recursos forem necessários, é possível fazer a transição para o BaseChatModel descrito abaixo.

A herança de SimpleChatModel requer a implementação da seguinte interface:

• Método _call - implementação de chamadas de API externas do modelo.

Adicionalmente, o seguinte pode ser especificado:

• Atributo _identifying_params - utilizado para registrar informações parametrizadas do modelo.

Opcional:

• Método _stream - utilizado para implementação de saída em streaming.

Modelo de Chat Base

A herança de BaseChatModel requer a implementação do método _generate e do atributo _llm_type. Opcionalmente, é possível implementar _stream, _agenerate, _astream e _identifying_params.

Exemplo de um Modelo de Chat Personalizado

Nesta seção, demonstraremos a implementação de código de um modelo de chat personalizado chamado CustomChatModelAdvanced, incluindo geração de resultados de chat, saída em streaming e implementação assíncrona de streaming.

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 um modelo de chat personalizado que retorna os primeiros `n` caracteres da última mensagem."""


    n: int
    """Parâmetro do modelo personalizado"""

    def _generate(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> ChatResult:
        """Aqui, a lógica de chamada do modelo é implementada, geralmente chamando a API de um modelo de terceiros e depois encapsulando o resultado retornado pela API em um formato que o langchain possa reconhecer.
        Explicação chave do parâmetro:
            messages: Uma lista de prompts compostos por mensagens
        """
        last_message = messages[-1]
        tokens = last_message.content[: self.n]
        message = AIMessage(content=tokens)
        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]:
        """Implementação de saída em streaming do modelo, similar ao método _generate, mas com processamento de saída em streaming."""
        last_message = messages[-1]
        tokens = last_message.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]:
        """Versão assíncrona da implementação do método `stream`"""
        result = 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 result:
            yield chunk

    @property
    def _llm_type(self) -> str:
        """Retorna a tag do modelo personalizado"""
        return "modelo-de-chat-eco-avançado"

    @property
    def _identifying_params(self) -> Dict[str, Any]:
        """Retorna informações de depuração personalizadas"""
        return {"n": self.n}

Testando o Modelo de Chat Personalizado

Vamos testar o modelo de chat, incluindo o uso dos métodos invoke, batch, stream e implementação assíncrona de streaming.

modelo = CustomChatModelAdvanced(n=3)

modelo.invoke([HumanMessage(content="Olá!")])

modelo.invoke("Olá")

modelo.batch(["Olá", "adeus"])

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", versão="v1"):
    print(evento)