مدل‌های سفارشی LangChain

سفارشی‌سازی LLM

در حال حاضر در حوزه مدل‌های هوش مصنوعی، تنوع زیادی از مدل‌ها و ادغام‌های رسمی LangChain وجود دارد که همهٔ مدل‌ها را نمی‌پوشانند. گاهی اوقات ممکن است نیاز داشته باشید یک مدل را سفارشی کنید و آن را در چارچوب LangChain ادغام کنید.

در این فصل، چگونگی ایجاد یک بسته‌بندی LLM سفارشی را معرفی می‌کنیم، تا برای شما امکان استفاده از مدل خود یا مدل‌هایی که توسط LangChain پشتیبانی نمی‌شوند، فراهم شود.

در LangChain، اگر می‌خواهید از یک LLM خود یا یک بسته‌بندی متفاوت از آنی که توسط LangChain پشتیبانی می‌شود استفاده کنید، می‌توانید یک بسته‌بندی LLM سفارشی ایجاد کنید. یک LLM سفارشی تنها باید دو متد لازم را اجرا کند:

• یک متد _call که یک رشته را به عنوان ورودی می‌گیرد، تعدادی کلمهٔ اختیاری به عنوان متوقف ورودی و یک رشته را برمی‌گرداند که اجرای مدل را در متد _call ایجاد می‌کند.
• یک ویژگی _llm_type که یک رشته را برگرداند که نام مدل را نمایش دهد و تنها برای اهداف ورود و خروج استفاده می‌شود.

به علاوه از متدهای لازم، یک LLM سفارشی می‌تواند یک متد اختیاری را نیز پیاده سازی کند:

• یک ویژگی _identifying_params که برای کمک به چاپ کلاس استفاده می‌شود. باید یک دیکشنری را برگرداند.

اجرای یک LLM سفارشی ساده

بیایید یک LLM سفارشی بسیار ساده پیاده سازی کنیم که تنها اولین n کاراکتر ورودی را برمی‌گرداند.

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 را سفارشی کنیم.

ورودی و خروجی پیام

در مدل گفتگو، پیام‌ها محور ورودی و خروجی هستند. یک پیام، محتوایی است که توسط کاربر وارد می‌شود و یا پاسخ تولیدی توسط مدل.

پیام‌ها

مدل گفتگو پیام‌ها را به عنوان ورودی می‌گیرد و سپس یک یا چند پیام را به عنوان خروجی تولید می‌کند. در LangChain، چندین نوع پیام داخلی وجود دارد که شامل:

• SystemMessage: برای مقدمهٔ رفتار هوش مصنوعی به‌کار می‌رود، معمولاً به عنوان اولین پیام در یک سری پیام ورودی.
• HumanMessage: نمایانگر تعامل کاربر با مدل گفتگو است.
• AIMessage: نمایانگر پیام‌هایی از مدل گفتگو است که می‌تواند متن یا درخواست فراخوانی ابزار باشد.
• FunctionMessage / ToolMessage: برای گذراندن نتیجهٔ فراخوانی یک ابزار به مدل استفاده می‌شود.

استفاده از این انواع پیام می‌تواند براساس نیازهای خاص گسترش یابد و سفارشی شود، مانند تنظیم براساس پارامترهای function و tool OpenAI.

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)