LangChain Benutzerdefinierte Modelle

Anpassen von LLM

Im aktuellen Bereich der KI-Modelle gibt es eine Vielzahl von Modellen, und die offizielle Integration von LangChain deckt nicht alle Modelle ab. Manchmal müssen Sie möglicherweise ein Modell anpassen und in das LangChain-Framework integrieren.

In diesem Kapitel wird erläutert, wie Sie einen benutzerdefinierten LLM-Wrapper erstellen können, um es Ihnen zu erleichtern, Ihr eigenes Modell oder Modelle, die nicht von LangChain unterstützt werden, zu verwenden.

In LangChain können Sie, wenn Sie Ihr eigenes LLM oder einen anderen Wrapper als den von LangChain unterstützten verwenden möchten, einen benutzerdefinierten LLM-Wrapper erstellen. Ein benutzerdefinierter LLM muss nur zwei erforderliche Methoden implementieren:

• Eine _call-Methode, die einen String als Eingabe und optionale Stoppwörter entgegennimmt und einen String zurückgibt, der die Modellaufruf-Implementierung in der _call-Methode darstellt.
• Ein _llm_type-Attribut, das einen String zurückgibt, der den Modellnamen repräsentiert und ausschließlich für Protokollierungszwecke verwendet wird.

Neben den erforderlichen Methoden kann ein benutzerdefiniertes LLM auch eine optionale Methode implementieren:

• Ein _identifying_params-Attribut, das verwendet wird, um die Klasse auszudrucken. Es sollte ein Wörterbuch zurückgeben.

Implementieren eines einfachen benutzerdefinierten LLM

Lassen Sie uns einen sehr einfachen benutzerdefinierten LLM implementieren, der nur die ersten n Zeichen der Eingabe zurückgibt.

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

    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("Stopp-Wörter sind nicht erlaubt.")
        return prompt[: self.n]

    @property
    def _identifying_params(self) -> Mapping[str, Any]:
        """Die Identifikationsparameter abrufen."""
        return {"n": self.n}

Jetzt können wir dieses benutzerdefinierte LLM genauso wie jedes andere LLM verwenden.

Verwenden des benutzerdefinierten LLM

Wir können das benutzerdefinierte LLM-Objekt instanziieren und verwenden, um zu demonstrieren, wie man das benutzerdefinierte LLM aufruft und die Ausgabe individuell anpasst.

llm = CustomLLM(n=10)

llm.invoke("Das ist eine tolle Sache")

'Das ist ein'

Außerdem können wir das LLM drucken und seine individualisierte Ausgabe anzeigen.

print(llm)

CustomLLM
Parameter: {'n': 10}

Anpassen des Chat-Modells

Hier erklären wir, wie Sie das Chat-Modell von LangChain anpassen können.

Nachrichteneingabe und -ausgabe

Im Chat-Modell stehen Nachrichten im Mittelpunkt von Eingabe und Ausgabe. Eine Nachricht ist der Inhalt, den der Benutzer eingibt, und die Antwort, die vom Modell generiert wird.

Nachrichten

Das Chat-Modell nimmt Nachrichten als Eingabe entgegen und generiert dann eine oder mehrere Nachrichten als Ausgabe. In LangChain gibt es verschiedene integrierte Nachrichtentypen, darunter:

• SystemMessage: Wird zur Initialisierung des KI-Verhaltens verwendet, normalerweise als erste Nachricht in einer Reihe von Eingabe-Nachrichten.
• HumanMessage: Stellt die Interaktion des Benutzers mit dem Chat-Modell dar.
• AIMessage: Stellt Nachrichten des Chat-Modells dar, die Text oder Tool-Aufrufanforderungen sein können.
• FunctionMessage / ToolMessage: Wird verwendet, um das Ergebnis eines Tool-Aufrufs an das Modell zurückzugeben.

Die Verwendung dieser Nachrichtentypen kann entsprechend spezifischer Anforderungen erweitert und angepasst werden, z. B. durch Anpassen basierend auf den Parametern function und tool von OpenAI.

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

Fluent-Varianten

Alle Chat-Nachrichten haben eine fließende Variante mit Chunk in ihren Namen.

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

Diese Chunk werden beim Streamen des Chat-Modells verwendet und jeder von ihnen definiert ein akkumulatives Attribut!

Beispiel

AIMessageChunk(content="Hallo") + AIMessageChunk(content=" Welt!")

Gibt zurück

AIMessageChunk(content='Hallo Welt!')

Einfaches Chat-Modell

Die Vererbung von SimpleChatModel ermöglicht die schnelle Implementierung eines einfachen Chat-Modells.

Obwohl es möglicherweise nicht alle für ein Chat-Modell erforderlichen Funktionen enthält, ermöglicht es eine schnelle Implementierung. Wenn mehr Funktionen erforderlich sind, ist ein Übergang zum unten beschriebenen BaseChatModell möglich.

Die Vererbung von SimpleChatModel erfordert die Implementierung des folgenden Interfaces:

• _call-Methode - Implementierung externer Modell-API-Aufrufe.

Zusätzlich können folgende Punkte spezifiziert werden:

• _identifying_params-Attribut - zur Aufzeichnung modellspezifischer Informationen.

Optional:

• _stream-Methode - zur Implementierung von Streaming-Ausgabe.

Basis-Chat-Modell

Die Vererbung von BaseChatModel erfordert die Implementierung der _generate-Methode und des _llm_type-Attributs. Optional ist auch die Implementierung von _stream, _agenerate, _astream und _identifying_params möglich.

Beispiel eines benutzerdefinierten Chat-Modells

In diesem Abschnitt zeigen wir die Code-Implementierung eines benutzerdefinierten Chat-Modells namens CustomChatModelAdvanced, einschließlich der Generierung von Chat-Ergebnissen, der Streaming-Ausgabe und der asynchronen Stream-Implementierung.

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):
    """Implementieren Sie ein benutzerdefiniertes Chat-Modell, das die ersten `n` Zeichen der letzten Nachricht zurückgibt."""


    n: int
    """Benutzerdefinierte Modellparameter"""

    def _generate(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> ChatResult:
        """Hier wird die Modellaufruflogik implementiert, normalerweise durch Aufrufen der API eines Drittanbietermodells und dann das Ergebnis, das von der API zurückgegeben wird, in ein Format zu verpacken, das von Langchain erkannt werden kann.
        Schlüsselparametererklärung:
            messages: Eine Liste von Prompts, die aus Nachrichten zusammengesetzt sind
        """
        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]:
        """Implementierung von Modell-Streaming-Ausgabe, ähnlich der Methode `_generate`, aber mit der Verarbeitung von Streaming-Ausgabe."""
        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]:
        """Asynchrone Version der `stream`-Methode Implementierung"""
        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:
        """Gibt den Tag des benutzerdefinierten Modells zurück"""
        return "echoing-chat-model-advanced"

    @property
    def _identifying_params(self) -> Dict[str, Any]:
        """Gibt benutzerdefinierte Debug-Informationen zurück"""
        return {"n": self.n}

Testen des benutzerdefinierten Chat-Modells

Lassen Sie uns das Chat-Modell testen, einschließlich der Verwendung der invoke, batch, stream-Methoden und der asynchronen Stream-Implementierung.

model = CustomChatModelAdvanced(n=3)

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

model.invoke("Hallo")

model.batch(["Hallo", "Auf Wiedersehen"])

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

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

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