LangChain Custom Models

تخصيص LLM

في مجال نماذج الذكاء الاصطناعي الحالي، هناك مجموعة كبيرة من النماذج، والاندماج الرسمي لـ LangChain لا يغطي جميع النماذج. في بعض الأحيان قد تحتاج إلى تخصيص نموذج ودمجه في إطار LangChain الخاص بك.

هذا الفصل سيقدم كيفية إنشاء غلاف LLM مخصص، مما يجعل من السهل بالنسبة لك استخدام نموذج خاص بك أو نماذج غير مدعمة من قبل LangChain.

في LangChain، إذا كنت ترغب في استخدام LLM الخاص بك أو غلاف مختلف عن الذي يدعمه LangChain، يمكنك إنشاء غلاف LLM مخصص. تحتاج LLM المخصص إلى تنفيذ طريقتين مطلوبتين فقط:

• طريقة _call، التي تأخذ سلسلة كإدخال، وبعض الكلمات الأمان الاختيارية، وتُرجع سلسلة، مطبقة استدعاء النموذج في طريقة _call.
• سمة _llm_type، تُرجع سلسلة تمثل اسم النموذج، تُستخدم فقط لأغراض السجلات.

بالإضافة إلى الطرق المطلوبة، يمكن لـ LLM المخصصة أيضًا تنفيذ طريقة اختيارية:

• سمة _identifying_params، تُستخدم للمساعدة في طباعة الفئة. يجب أن تُرجع قاموسًا.

تنفيذ LLM مخصص بسيط

لنقم بتنفيذ 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("لا يُسمح بمعاملات التوقف.")
        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("هذا شيء رائع جدا")

'هذا شيء '

يمكننا أيضًا طباعة LLM ومشاهدة إخراج الطباعة المخصص له.

print(llm)

CustomLLM
Params: {'n': 10}

تخصيص نموذج المحادثة

سنقوم هنا بشرح كيفية تخصيص نموذج الدردشة في LangChain.

إدخال وإخراج الرسائل

في نموذج الدردشة، تكون الرسائل محور الإدخال والإخراج. الرسالة هي المحتوى الذي يُدخله المستخدم والرد الذي يتم إنشاؤه من قبل النموذج.

الرسائل

يتخذ نموذج الدردشة الرسائل كإدخال ثم يُنشئ رسالة واحدة أو أكثر كإخراج. في LangChain، هناك عدة أنواع مدمجة للرسائل، بما في ذلك:

• SystemMessage: تُستخدم لتهيئة سلوك الذكاء الاصطناعي، عادةً كالرسالة الأولى في سلسلة من رسائل الإدخال.
• HumanMessage: تمثل تفاعل المستخدم مع نموذج الدردشة.
• AIMessage: تمثل رسائل من نموذج الدردشة، والتي يمكن أن تكون نصًا أو طلبات استدعاء أداة.
• FunctionMessage / ToolMessage: تُستخدم لتمرير نتيجة استدعاء أداة مرة أخرى إلى النموذج.

يمكن توسيع استخدام هذه الأنواع من الرسائل وتخصيصها وفقًا للمتطلبات الخاصة، مثل التكيف بمعلمات 'الدالة' و 'الأداة' من 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="مرحبا") + AIMessageChunk(content=" يا عالم!")

الناتج

AIMessageChunk(content='مرحبا يا عالم!')

نموذج الدردشة البسيط

التوريث من SimpleChatModel يسمح بتنفيذ نموذج دردشة بسيط بسرعة.

على الرغم من أنه قد لا يتضمن جميع الوظائف المطلوبة لنموذج الدردشة، إلا أنه يوفر تنفيذًا سريعًا. إذا كانت هناك حاجة إلى مزيد من الميزات، يمكن الانتقال إلى BaseChatModel الموضح أدناه.

يتطلب التوريث من SimpleChatModel تنفيذ واجهة التالية:

• طريقة _call - تنفيذ لمكالمات واجهة برمجة التطبيقات الخارجية.

بالإضافة إلى ذلك، يمكن تحديد ما يلي:

• السمة _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) لنموذج طرف ثالث، ثم تجميع النتيجة المُرَجَعة بواسطة الواجهة إلى شكل يمكن للنظام أن يعترف به.
        تفسير المعلمة الرئيسية:
            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="مرحبًا!")])

model.invoke("مرحبا")

model.batch(["مرحبا", "وداعا"])

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

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

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