LangChain Custom Models

কাস্টমাইজিং LLM

বর্তমান AI মডেল ডোমেইনে, মডেলের একটি বৃহৎ পরিমাণ রয়েছে, এবং LangChain এর অফিসিয়াল ইন্টিগ্রেশন সব মডেল কাভার করে না। কিছুসময় আপনি নিজের মডেল তৈরি করতে চাইতে পারেন এবং তা ল্যাংচেইন ফ্রেমওয়ার্কে ইন্টিগ্রেট করতে হতে পারে।

এই অধ্যায়টি আপনাকে একটি কাস্টম LLM র্হাতার তৈরী করার উপযুক্ত করে এনার সুবিধা প্রদান করবে, আপনি যে কোনও আপনার নিজের মডেল বা LangChain দ্বারা সমর্থিত নয় মডেল ব্যবহার করতে পারেন।

ল্যাংচেইনে, আপনি আপনার নিজের LLM বা LangChain দ্বারা সমর্থিত নয় একটি ভিন্ন wrapper ব্যবহার করতে চাইলে একটি কাস্টম LLM wrapper তৈরী করতে পারেন। একটি কাস্টম 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("স্টপ কোয়ার্ড অনুমতি দেওয়া হয়নি।")
        return prompt[: self.n]

    @property
    def _identifying_params(self) -> Mapping[str, Any]:
        """সনাক্ত প্যারামিটার পেতে।"""
        return {"n": self.n}

এখন আমরা এই কাস্টম LLM প্রত্যাহার যেমন যে কোনও অন্য এলএলএম এর মতো ব্যবহার করতে পারি।

কাস্টম LLM ব্যবহার করা

আমরা কাস্টম LLM অবজেক্ট তৈরি করতে এবং ব্যবহার করতে পারি, যাতে করে কাস্টম LLM আহ্বান করতে ও মূল্যায়ন করার সুযোগ পাই।

llm = CustomLLM(n=10)

llm.invoke("এটা একটি ফুবার জিনিস")

'এটা একটি '

আমরা আগামীতে প্রিন্ট করতে এবং তার কাস্টমাইজড প্রিন্ট আউটপুট দেখতে পারি।

print(llm)

CustomLLM
প্যারামিটার: {'n': 10}

চ্যাট মডেল কাস্টমাইজ করা

এখানে আমরা আলোচনা করবো কীভাবে LangChain এর চ্যাট মডেল কাস্টমাইজ করা যায়।

মেসেজ ইনপুট এবং আউটপুট

চ্যাট মডেলে, মেসেজগুলি ইনপুট এবং আউটপুটের জন্য কেন্দ্র করা হয়। একটি মেসেজ হ'ল ব্যবহারকারী দ্বারা এন্টার করা মেসেজের সামগ্রী এবং মডেল দ্বারা তৈরি উত্তর।

মেসেজ

চ্যাট মডেল মেসেজগুলি ইনপুট হিসেবে নেয় এবং তারপর আউটপুট হিসেবে এক বা একাধিক মেসেজ তৈরি করে। 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="হ্যালো") + AIMessageChunk(content=" বিশ্ব!")

ফিরেঃ

AIMessageChunk(content='হ্যালো বিশ্ব!')

সাধারণ চ্যাট মডেল

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 কল করার মাধ্যমে এবং তারপরে একটি ফরম্যাটে রূপান্তর করার মাধ্যমে, 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]:
        """মডেল স্ট্রীমিং আউটপুট কার্যকর করার উদ্দেশ্যে এর কার্যের অনুমতি দেয়া হয়, মধ্যে স্ট্রীমিং আউটপুট প্রসেসিং করা হয়।"""
        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]:
        """স্ট্রিম মেথড অমলের অসিঞ্চিত সংস্করণ"""
        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 "ইকোয়িং-চ্যাট-মডেল-অগ্রিম"

    @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)