लैंगचेन कस्टम मॉडल

LLM कस्टमाइज़ करना

वर्तमान AI मॉडल डोमेन में, मॉडलों की एक बड़ी विविधता है, और LangChain की आधिकारिक एकीकरण सभी मॉडलों को कवर नहीं करता। कभी-कभी आपको अपना मॉडल कस्टमाइज़ करके और इसे LangChain के फ़्रेमवर्क में एकीकृत करने की आवश्यकता हो सकती है।

इस अध्याय में यह समझाया जाएगा कि एक Custom LLM wrapper कैसे बनाया जाए, जिससे आप अपने खुद के मॉडल का उपयोग कर सकें या LangChain द्वारा समर्थित न होने वाले मॉडल्स को फ़्रेमवर्क में शामिल कर सकें।

LangChain में, अगर आप अपना खुद का LLM या LangChain द्वारा समर्थित न होने वाले wrapper का उपयोग करना चाहते हैं, तो आप एक Custom LLM wrapper बना सकते हैं। एक Custom LLM को केवल दो आवश्यक methods को implement करने की आवश्यकता होती है:

• एक _call method, जिसमें एक string को input के रूप में लेता है, कुछ optional stop words को, और एक string को return करता है, _call method में मॉडल invoke को implement करता है।
• एक _llm_type attribute, जो केवल logging के लिए उपयोग होने वाले मॉडल का नाम दर्शाता है, वापसी attributes को return करता है।

आवश्यक methods के अलावा, एक Custom LLM एक optional method को implement कर सकता है:

• एक _identifying_params attribute, जो क्लास को प्रिंट करने में मदद करने के लिए उपयोग होता है। यह एक dictionary को return करता है।

एक सरलतम Custom LLM को implement करना

चलो, हम एक बहुत ही सरल Custom LLM को implement करते हैं जो केवल input के पहले n characters को वापस लौटाता है।

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}

अब हम इस custom LLM को किसी भी अन्य LLM की तरह उपयोग कर सकते हैं।

Custom LLM का उपयोग करना

हम custom LLM object को instantiate कर सकते हैं और उसका उपयोग करके दिखा सकते हैं, custom LLM को invoke कैसे किया जाता है और प्रिंट output को customize किया जाता है।

llm = CustomLLM(n=10)

llm.invoke("यह एक फूबार चीज़ है")

'यह एक फूबार'

हम चाहें तो LLM को print कर सकते हैं और उसके customize print output को देख सकते हैं।

print(llm)

CustomLLM
Params: {'n': 10}

चैट मॉडल कस्टमाइज़ करना

यहाँ हम समझाएँगे कि LangChain के चैट मॉडल को कस्टमाइज़ कैसे किया जाए।

संदेश Input और Output

चैट मॉडल में, संदेश input और output का मुद्दा होता है। एक संदेश उपयोगकर्ता द्वारा दर्ज किए गए सामग्री और मॉडल द्वारा बनाए गए प्रतिक्रिया को दर्शाता है।

संदेश

चैट मॉडल मैसेज को input के रूप में लेता है और फिर एक या एक से अधिक संदेश को output के रूप में उत्पन्न करता है। LangChain में, कई built-in मैसेज प्रकार होते हैं, जिनमें शामिल हैं:

• SystemMessage: AI व्यवहार की शुरुआत के लिए उपयोग किया जाता है, सामान्यतः input मैसेजों की एक श्रृंखला में पहला मैसेज।
• HumanMessage: चैट मॉडल के साथ उपयोगकर्ता के interaction को दर्शाता है।
• AIMessage: चैट मॉडल से मैसेज का प्रतिनिधित्व करता है, जो टेक्स्ट या टूल invoke रिक्वेस्ट हो सकता है।
• FunctionMessage / ToolMessage: टूल invoke के परिणाम को मॉडल को वापस देने के लिए उपयोग किया जाता है।

इन मैसेज प्रकारों के उपयोग को विस्तारित और कस्टमाइज़ किया जा सकता है, जैसे कि OpenAI के function और tool पैरामीटर्स पर आधारित समायोजन करना।

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

फ्ल्यूएंट वेरिएंट्स

सभी चैट मैसेज के पास उनके नामों में Chunk के साथ एक fluent variant होता है।

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

ये Chunk संदेश मॉडल को स्ट्रीम करते समय उपयोग किए जाते हैं, और प्रत्येक इनमें एक accumulative attribute को define करता है!

उदाहरण

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 को कॉल करके, और फिर 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)