Menyesuaikan LLM

Di dalam domain model AI saat ini, terdapat beragam model, dan integrasi resmi LangChain tidak mencakup semua model. Terkadang Anda mungkin perlu menyesuaikan model dan mengintegrasikannya ke dalam framework LangChain.

Bagian ini akan memperkenalkan cara membuat pembungkus LLM kustom, sehingga memudahkan Anda untuk menggunakan model Anda sendiri atau model yang tidak didukung oleh LangChain.

Di dalam LangChain, jika Anda ingin menggunakan LLM Anda sendiri atau pembungkus yang berbeda dari yang didukung oleh LangChain, Anda dapat membuat pembungkus LLM kustom. Sebuah LLM kustom hanya perlu mengimplementasikan dua metode yang diperlukan:

  • Metode _call, yang mengambil masukan berupa string, beberapa kata berhenti opsional, dan mengembalikan string, mengimplementasikan panggilan model dalam metode _call.
  • Atribut _llm_type, yang mengembalikan string yang mewakili nama model, digunakan semata-mata untuk tujuan pencatatan.

Selain metode yang diperlukan, sebuah LLM kustom juga dapat mengimplementasikan metode opsional:

  • Atribut _identifying_params, digunakan untuk membantu mencetak kelas. Atribut ini harus mengembalikan kamus.

Implementasi LLM kustom sederhana

Mari kita implementasikan LLM kustom yang sangat sederhana yang hanya mengembalikan karakter pertama n dari masukan.

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

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

    @property
    def _identifying_params(self) -> Mapping[str, Any]:
        """Dapatkan parameter identifikasi."""
        return {"n": self.n}

Kini kita dapat menggunakan LLM kustom ini seperti halnya LLM lainnya.

Menggunakan LLM kustom

Kita dapat membuat objek LLM kustom dan menggunakan objek LLM kustom tersebut, menunjukkan bagaimana memanggil LLM kustom dan menyesuaikan keluaran cetak.

llm = CustomLLM(n=10)

llm.invoke("Ini adalah hal yang menyenangkan")

'Ini adalah h'

Kita juga dapat mencetak LLM dan melihat keluaran cetak kustomnya.

print(llm)

CustomLLM
Params: {'n': 10}

Menyesuaikan Model Obrolan

Di sini kita akan menjelaskan cara menyesuaikan model obrolan LangChain.

Masukan dan Keluaran Pesan

Di dalam model obrolan, pesan adalah fokus dari masukan dan keluaran. Sebuah pesan adalah konten yang dimasukkan oleh pengguna dan tanggapan yang dihasilkan oleh model.

Pesan

Model obrolan mengambil pesan sebagai masukan dan kemudian menghasilkan satu atau lebih pesan sebagai keluaran. Di dalam LangChain, terdapat beberapa jenis pesan bawaan, termasuk:

  • SystemMessage: Digunakan untuk menginisialisasi perilaku AI, biasanya sebagai pesan pertama dalam serangkaian pesan masukan.
  • HumanMessage: Mewakili interaksi pengguna dengan model obrolan.
  • AIMessage: Mewakili pesan dari model obrolan, yang dapat berupa teks atau permintaan panggilan alat.
  • FunctionMessage / ToolMessage: Digunakan untuk meneruskan hasil panggilan alat kembali ke model.

Penggunaan jenis pesan ini bisa diperluas dan disesuaikan sesuai dengan kebutuhan spesifik, seperti penyesuaian berdasarkan parameter function dan tool dari OpenAI.

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

Varian Yang Mengalir

Semua pesan obrolan memiliki varian yang mengalir dengan Chunk di dalam namanya.

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

Chunk ini digunakan saat menyiarkan model obrolan, dan masing-masing dari mereka mendefinisikan atribut yang mengakumulasikan!

Contoh

AIMessageChunk(content="Halo") + AIMessageChunk(content=" dunia!")

Mengembalikan

AIMessageChunk(content='Halo dunia!')

Model Chat Sederhana

Dengan mewarisi dari SimpleChatModel memungkinkan untuk dengan cepat menerapkan model obrolan sederhana.

Meskipun mungkin tidak mencakup semua fungsionalitas yang dibutuhkan untuk model obrolan, ini memberikan implementasi yang cepat. Jika lebih banyak fitur diperlukan, transisi ke BaseChatModel yang dijelaskan di bawah ini memungkinkan.

Mewarisi dari SimpleChatModel memerlukan implementasi dari antarmuka berikut:

  • metode _call - implementasi panggilan API model eksternal.

Selain itu, yang berikut dapat ditentukan:

  • atribut _identifying_params - digunakan untuk mencatat informasi parameter model.

Opsional:

  • metode _stream - digunakan untuk mengimplementasikan output streaming.

Model Obrolan Dasar

Mewarisi dari BaseChatModel memerlukan implementasi dari metode _generate dan atribut _llm_type. Opsionalnya, implementasi dari _stream, _agenerate, _astream, dan _identifying_params juga memungkinkan.

Contoh Model Obrolan Kustom

Pada bagian ini, kami akan menunjukkan implementasi kode dari model obrolan kustom yang disebut CustomChatModelAdvanced, termasuk menghasilkan hasil obrolan, output streaming, dan implementasi streaming asinkron.

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):
    """Menerapkan model obrolan kustom yang mengembalikan karakter pertama `n` dari pesan terakhir."""


    n: int
    """Parameter model kustom"""

    def _generate(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> ChatResult:
        """Di sini, logika pemanggilan model diimplementasikan, biasanya dengan memanggil API dari model pihak ketiga, dan kemudian mengemas hasil yang dikembalikan oleh API ke dalam format yang dapat dikenali oleh langchain.
        Penjelasan parameter kunci:
            messages: Sebuah daftar dari rangsangan terdiri dari pesan-pesan
        """
        pesan_terakhir = messages[-1]
        token = pesan_terakhir.content[: self.n]
        pesan = AIMessage(content=token)
        hasil_generasi = ChatGeneration(message=pesan)
        return ChatResult(generations=[hasil_generasi])

    def _stream(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> Iterator[ChatGenerationChunk]:
        """Implementasi output streaming model, mirip dengan metode _generate, tetapi dengan pengolahan output streaming."""
        pesan_terakhir = messages[-1]
        token = pesan_terakhir.content[: self.n]

        for token in token:
            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 asinkron dari implementasi metode `stream`"""
        hasil = 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 hasil:
            yield chunk

    @property
    def _llm_type(self) -> str:
        """Mengembalikan tag dari model kustom"""
        return "echoing-chat-model-advanced"

    @property
    def _identifying_params(self) -> Dict[str, Any]:
        """Mengembalikan informasi debug kustom"""
        return {"n": self.n}

Pengujian Model Obrolan Kustom

Mari uji model obrolan, termasuk menggunakan metode invoke, batch, stream, dan implementasi stream asinkron.

model = CustomChatModelAdvanced(n=3)

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

model.invoke("Halo")

model.batch(["Halo", "Selamat tinggal"])

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

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

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