LLMのカスタマイズ

現在のAIモデルの領域には、さまざまなモデルがあり、LangChainの公式統合ではすべてのモデルをカバーしていません。時には独自のモデルをカスタマイズしてLangChainフレームワークに統合する必要があることがあります。

この章では、独自のLLMラッパーを作成して、独自のモデルまたはLangChainでサポートされていないモデルを使用できるようにする方法について紹介します。

LangChainでは、独自のLLMまたはLangChainでサポートされているラッパーとは異なるラッパーを使用したい場合は、カスタムLLMラッパーを作成できます。独自のLLMは、2つの必須メソッドを実装するだけです。

  • _callメソッド：文字列とオプションのストップワードを入力として受け取り、文字列を返すメソッドで、_callメソッド内でのモデルの実行を実装します。
  • _llm_type属性：ログ記録目的だけで使用されるモデル名を表す文字列を返します。

必須のメソッドに加えて、カスタムLLMはオプションのメソッドを実装することもできます。

  • _identifying_params属性：クラスを表示するのに役立つ属性であり、辞書を返します。

シンプルなカスタムLLMの実装

入力の最初のn文字だけを返す非常にシンプルなカスタムLLMを実装してみましょう。

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]:
        """識別パラメータを取得します。"""
        return {"n": self.n}

これで、このカスタムLLMを他のLLMと同様に使用できるようになります。

カスタムLLMの使用

カスタムLLMオブジェクトをインスタンス化して使用し、カスタムLLMを呼び出して出力をカスタマイズする方法を示します。

llm = CustomLLM(n=10)

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

'This is a '

また、LLMを表示してそのカスタム化された出力を表示することもできます。

print(llm)

CustomLLM
Params: {'n': 10}

チャットモデルのカスタマイズ

そしてここでは、LangChainのチャットモデルをカスタマイズする方法について説明します。

メッセージの入力と出力

チャットモデルでは、メッセージが入力と出力の焦点です。メッセージはユーザーによって入力されたコンテンツと、モデルによって生成された応答です。

メッセージ

チャットモデルでは、メッセージを入力として受け取り、1つ以上のメッセージを出力します。LangChainでは、次のような組み込みのメッセージタイプがあります。

  • SystemMessage：AIの動作を初期化するために使用され、通常は入力メッセージの最初のメッセージとして使われます。
  • HumanMessage：チャットモデルによるユーザーとのやり取りを表します。
  • AIMessage：チャットモデルからのメッセージを表し、テキストまたはツールの呼び出し要求が含まれることがあります。
  • FunctionMessage / ToolMessage：ツールの呼び出しの結果をモデルに戻すために使用されます。

これらのメッセージタイプの使用は、特定の要件に応じて拡張およびカスタマイズすることができ、OpenAIのfunctionとtoolパラメータに基づいて調整することができます。

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

フルーエントバリアント

すべてのチャットメッセージには、名前にChunkが付いたフルーエントバリアントがあります。

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

これらのChunkは、チャットモデルをストリーミングする際に使用され、それぞれが蓄積属性を定義しています！

例

AIMessageChunk(content="Hello") + AIMessageChunk(content=" world!")

戻り値

AIMessageChunk(content='Hello world!')

シンプルチャットモデル

SimpleChatModel を継承すると、シンプルなチャットモデルを迅速に実装することができます。

全てのチャットモデルに必要な機能を組み込むわけではありませんが、迅速な実装が可能です。より多くの機能が必要な場合は、以下で説明する BaseChatModel への移行が可能です。

SimpleChatModel を継承すると、次のインタフェースの実装が必要です。

  • _call メソッド - 外部モデル API 呼び出しの実装。

さらに、次のように指定することができます。

  • _identifying_params 属性 - モデルのパラメータ化された情報を記録するために使用されます。

オプション:

  • _stream メソッド - ストリーミング出力の実装に使用されます。

ベースチャットモデル

BaseChatModel を継承すると、_generate メソッドと _llm_type 属性の実装が必要です。オプションとして、_stream、_agenerate、_astream、_identifying_params の実装が可能です。

カスタムチャットモデルの例

このセクションでは、CustomChatModelAdvanced というカスタムチャットモデルのコード実装例を示します。チャット結果の生成、ストリーミング出力、非同期ストリーム実装が含まれます。

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):
    """直前のメッセージの最初の `n` 文字を返すカスタムチャットモデルを実装"""


    n: int
    """カスタムモデルのパラメータ"""

    def _generate(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> ChatResult:
        """ここでは、通常、第三者モデルの API を呼び出してロジックを実装し、その後 API が返した結果を langchain が認識できる形式にカプセル化します。
        キーとなるパラメータの説明:
            messages: メッセージで構成されたプロンプトのリスト
        """
        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]:
        """モデルのストリーミング出力の実装。_generate メソッドと類似していますが、ストリーミング出力処理が行われます。"""
        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]:
        """`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:
        """カスタムモデルのタグを返す"""
        return "echoing-chat-model-advanced"

    @property
    def _identifying_params(self) -> Dict[str, Any]:
        """カスタムのデバッグ情報を返す"""
        return {"n": self.n}

カスタムチャットモデルのテスト

invoke、batch、stream メソッドの使用、および非同期ストリーム実装を含む、チャットモデルのテストを行いましょう。

model = CustomChatModelAdvanced(n=3)

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

model.invoke("hello")

model.batch(["hello", "goodbye"])

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

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

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