■ BaseChatModel 클래스를 사용해 사용자 정의 채팅 모델을 만드는 방법을 보여준다.
▶ main.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 |
from typing import Optional from typing import List from typing import Any from typing import Iterator from typing import Dict from langchain_core.language_models import BaseChatModel from langchain_core.messages import BaseMessage from langchain_core.callbacks import CallbackManagerForLLMRun from langchain_core.messages import AIMessage from langchain_core.outputs import ChatGeneration from langchain_core.outputs import ChatResult from langchain_core.outputs import ChatGenerationChunk from langchain_core.messages import AIMessageChunk from langchain_core.messages import HumanMessage class CustomChatModel(BaseChatModel): """A custom chat model that echoes the first `n` characters of the input. When contributing an implementation to LangChain, carefully document the model including the initialization parameters, include an example of how to initialize the model and include any relevant links to the underlying models documentation or API. Example: .. code-block:: python model = CustomChatModel(n = 2) result = model.invoke([HumanMessage(content = "hello")]) result = model.batch([[HumanMessage(content = "hello")], [HumanMessage(content = "world")]]) """ modelName : str """The name of the model""" echoCharacterCount : int """The number of characters from the last message of the prompt to be echoed.""" def _generate( self, messageList : List[BaseMessage], stop : Optional[List[str]] = None, runManager : Optional[CallbackManagerForLLMRun] = None, **kwargs : Any ) -> ChatResult: """Override the _generate method to implement the chat model logic. This can be a call to an API, a call to a local model, or any other implementation that generates a response to the input prompt. Args: messages: the prompt composed of a list of messages. stop: a list of strings on which the model should stop generating. If generation stops due to a stop token, the stop token itself SHOULD BE INCLUDED as part of the output. This is not enforced across models right now, but it's a good practice to follow since it makes it much easier to parse the output of the model downstream and understand why generation stopped. run_manager: A run manager with callbacks for the LLM. """ # 메시지 목록에서 응답을 생성하려면 이를 실제 논리로 바꾸세요. lastMessage = messageList[-1] tokenList = lastMessage.content[: self.echoCharacterCount] message = AIMessage( content = tokenList, additional_kwargs = {}, # 추가 페이로드를 추가하는 데 사용됩니다(예 : 함수 호출 요청) response_metadata = {"time_in_seconds" : 3 } # 응답 메타데이터에 사용한다. ) generation = ChatGeneration(message = message) return ChatResult(generations = [generation]) def _stream( self, messageList : List[BaseMessage], stop : Optional[List[str]] = None, runManager : Optional[CallbackManagerForLLMRun] = None, **kwargs : Any ) -> Iterator[ChatGenerationChunk]: """Stream the output of the model. This method should be implemented if the model can generate output in a streaming fashion. If the model does not support streaming, do not implement it. In that case streaming requests will be automatically handled by the _generate method. Args: messages: the prompt composed of a list of messages. stop: a list of strings on which the model should stop generating. If generation stops due to a stop token, the stop token itself SHOULD BE INCLUDED as part of the output. This is not enforced across models right now, but it's a good practice to follow since it makes it much easier to parse the output of the model downstream and understand why generation stopped. run_manager: A run manager with callbacks for the LLM. """ lastMessage = messageList[-1] tokenList = lastMessage.content[: self.echoCharacterCount] for token in tokenList: chunk = ChatGenerationChunk(message = AIMessageChunk(content = token)) if runManager: # 최신 버전의 LangChain에서는 선택 사항이다. # on_llm_new_token이 자동으로 호출된다. runManager.on_llm_new_token(token, chunk = chunk) yield chunk # 다른 정보(예 : 응답 메타 데이터)를 추가해 보겠다. chunk = ChatGenerationChunk(message = AIMessageChunk(content = "", response_metadata = {"time_in_sec" : 3})) if runManager: # 최신 버전의 LangChain에서는 선택 사항이다. # on_llm_new_token이 자동으로 호출된다. runManager.on_llm_new_token(token, chunk = chunk) yield chunk @property def _llm_type(self) -> str: """Get the type of language model used by this chat model.""" return "CustomChatModel" @property def _identifying_params(self) -> Dict[str, Any]: """Return a dictionary of identifying parameters. This information is used by the LangChain callback system, which is used for tracing purposes make it possible to monitor LLMs. """ return { # 모델 이름을 통해 사용자는 LLM 모니터링 애플리케이션에서 사용자 정의 토큰 계산 규칙을 지정할 수 있다. # (예를 들어 LangSmith에서 사용자는 자신의 모델에 대한 토큰별 가격을 제공하고 특정 LLM에 대한 비용을 모니터링할 수 있다.) "model_name" : self.modelName } customChatModel = CustomChatModel(echoCharacterCount = 3, modelName = "customChatModel") responseAIMessage1 = customChatModel.invoke( [ HumanMessage(content = "hello!"), AIMessage(content = "Hi there human!"), HumanMessage(content = "Meow!") ] ) print(responseAIMessage1) """ content='Meo' response_metadata={'time_in_seconds': 3} id='run-b37efe20-1fc4-4f2b-9563-eed7c0995e32-0' """ responseAIMessage2 = customChatModel.invoke("hello") print(responseAIMessage2) """ content='hel' response_metadata={'time_in_seconds': 3} id='run-f478bbd4-d691-475a-923f-cfca8c2dd06c-0' """ responseAIMessage3 = customChatModel.invoke(["hello", "goodbye"]) print(responseAIMessage3) """ content='goo' response_metadata={'time_in_seconds': 3} id='run-f7e8a5eb-da30-40aa-ad0d-4b2e0daa0e79-0' """ for aiMessageChunk in customChatModel.stream("cat"): print(aiMessageChunk.content, end = "|") """ c|a|t|| """ |
▶ requirements.txt
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 |
aiohttp==3.9.5 aiosignal==1.3.1 annotated-types==0.7.0 async-timeout==4.0.3 attrs==23.2.0 certifi==2024.6.2 charset-normalizer==3.3.2 frozenlist==1.4.1 greenlet==3.0.3 idna==3.7 jsonpatch==1.33 jsonpointer==3.0.0 langchain==0.2.5 langchain-core==0.2.9 langchain-text-splitters==0.2.1 langsmith==0.1.82 multidict==6.0.5 numpy==1.26.4 orjson==3.10.5 packaging==24.1 pydantic==2.7.4 pydantic_core==2.18.4 PyYAML==6.0.1 requests==2.32.3 SQLAlchemy==2.0.31 tenacity==8.4.2 typing_extensions==4.12.2 urllib3==2.2.2 yarl==1.9.4 |
※ pip install langchain 명령을 실행했다.