Własne modele LangChain

Dostosowywanie LLM

W obecnej domenie modeli sztucznej inteligencji istnieje wiele różnych modeli, a oficjalna integracja LangChain nie obejmuje wszystkich modeli. Czasami możesz potrzebować dostosować model i zintegrować go z platformą LangChain.

Ten rozdział przedstawi, jak stworzyć niestandardowy opakowanie LLM, umożliwiając wygodne korzystanie z własnego modelu lub modeli, które nie są obsługiwane przez LangChain.

W LangChain, jeśli chcesz użyć własnego LLM lub innego opakowania niż to obsługiwane przez LangChain, możesz stworzyć niestandardowe opakowanie LLM. Niestandardowe LLM musi jedynie zaimplementować dwie wymagane metody:

• Metodę _call, która przyjmuje jako argument ciąg znaków, opcjonalne słowa przestankowe i zwraca ciąg znaków, implementując wywołanie modelu w metodzie _call.
• Atrybut _llm_type, zwracający ciąg znaków reprezentujący nazwę modelu, używany wyłącznie do celów logowania.

Oprócz wymaganych metod, niestandardowe LLM mogą również zaimplementować opcjonalną metodę:

• Atrybut _identifying_params, używany do pomocy w drukowaniu klasy. Powinien zwracać słownik.

Implementacja prostego niestandardowego LLM

Stwórzmy bardzo prosty niestandardowy LLM, który zwraca tylko pierwsze n znaków z wejścia.

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

    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("niedozwolone argumenty stop.")
        return prompt[: self.n]

    @property
    def _identifying_params(self) -> Mapping[str, Any]:
        """Pobierz parametry identyfikujące."""
        return {"n": self.n}

Teraz możemy używać tego niestandardowego LLM tak jak każdego innego LLM.

Używanie niestandardowego LLM

Możemy zainicjować i używać obiektu niestandardowego LLM, pokazując, jak wywołać niestandardowe LLM i dostosować wyjście w druku.

llm = CustomLLM(n=10)

llm.invoke("To jest jakiś przykład")

'To jest ja'

Możemy również wydrukować LLM i zobaczyć dostosowany wynik druku.

print(llm)

CustomLLM
Parametry: {'n': 10}

Dostosowywanie Modelu Czatu

Tutaj wyjaśnimy, jak dostosować model czatu w LangChain.

Wejście i Wyjście Wiadomości

W modelu czatu wiadomości stanowią główny punkt wejścia i wyjścia. Wiadomość to treść wprowadzana przez użytkownika oraz odpowiedź generowana przez model.

Wiadomości

Model czatu pobiera wiadomości jako dane wejściowe, a następnie generuje jedną lub więcej wiadomości jako dane wyjściowe. W LangChain istnieje kilka wbudowanych typów wiadomości, w tym:

• SystemMessage: Używane do inicjowania zachowania SI, zwykle jako pierwsza wiadomość w serii wiadomości wejściowych.
• HumanMessage: Reprezentuje interakcję użytkownika z modelem czatu.
• AIMessage: Reprezentuje wiadomości od modelu czatu, które mogą być tekstem lub żądaniami wywołania narzędzia.
• FunctionMessage / ToolMessage: Używane do przekazywania wyniku wywołania narzędzia z powrotem do modelu.

Wykorzystanie tych typów wiadomości można rozszerzyć i dostosować zgodnie z konkretnymi wymaganiami, takimi jak dostosowanie na podstawie parametrów „function” i „tool” OpenAI.

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

Wyraźne Warianty

Wszystkie wiadomości czatu mają warianty z atrybutem Chunk w ich nazwach.

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

Te „Chunk” są wykorzystywane podczas przesyłania strumieniowego modelu czatu i każdy z nich definiuje atrybut skumulowany!

Przykład

AIMessageChunk(content="Cześć") + AIMessageChunk(content=" świecie!")

Zwraca

AIMessageChunk(content='Cześć świecie!')

Prosty Model Czatu

Dziedziczenie po SimpleChatModel pozwala na szybką implementację prostego modelu czatu.

Chociaż może nie zawierać wszystkich funkcji potrzebnych do modelu czatu, umożliwia szybką implementację. Jeśli są wymagane dodatkowe funkcje, przejście do opisanego poniżej BaseChatModel jest możliwe.

Dziedziczenie po SimpleChatModel wymaga implementacji następującego interfejsu:

• Metoda _call - implementacja zewnętrznych wywołań interfejsu modelu.

Dodatkowo można zdefiniować następujące:

• Atrybut _identifying_params - wykorzystywany do rejestrowania zparametryzowanych informacji o modelu.

Opcjonalnie:

• Metoda _stream - służy do implementacji strumieniowego wyjścia.

Podstawowy Model Czatu

Dziedziczenie po BaseChatModel wymaga implementacji metody _generate i atrybutu _llm_type. Opcjonalnie można zaimplementować _stream, _agenerate, _astream i _identifying_params.

Przykład niestandardowego modelu czatu

W tej sekcji przedstawimy implementację kodu niestandardowego modelu czatu o nazwie CustomChatModelAdvanced, w tym generowanie wyników czatu, strumieniowe wyjście i implementację strumieni asynchronicznych.

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):
    """Zaimplementuj niestandardowy model czatu zwracający pierwsze `n` znaków ostatniej wiadomości."""


    n: int
    """Parametr niestandardowego modelu"""

    def _generate(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> ChatResult:
        """Tutaj zaimplementowana jest logika wywołań modelu, zwykle poprzez wywołanie interfejsu aplikacji zewnętrznego modelu, a następnie opakowanie wyniku zwróconego przez interfejs w format, który może rozpoznać langchain.
        Kluczowe wyjaśnienie parametrów:
            messages: Lista komunikatów składających się z wiadomości
        """
        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]:
        """Implementacja strumieniowego wyjścia modelu, podobna do metody _generate, ale z przetwarzaniem strumieniowym wyjścia."""
        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]:
        """Asynchroniczna wersja implementacji metody `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:
        """Zwróć tag niestandardowego modelu"""
        return "echoing-chat-model-advanced"

    @property
    def _identifying_params(self) -> Dict[str, Any]:
        """Zwróć niestandardowe informacje debugowania"""
        return {"n": self.n}

Testowanie niestandardowego modelu czatu

Sprawdźmy model czatu, w tym użycie metod invoke, batch, stream oraz asynchroniczną implementację strumieniową.

model = CustomChatModelAdvanced(n=3)

model.invoke([HumanMessage(content="cześć!")])

model.invoke("cześć")

model.batch(["cześć", "do widzenia"])

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

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

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