当前位置：首页 > ds >正文

如何基于langchain基类LLM自定义大模型

ds 2025/8/16 16:34:12

在langchain调用外部大模型时，往往会因为sk、ak参数不匹配犯难。

如果能自定义一个LLM，对这些外部模型进行封装，就可以采用熟悉方式使用这些外部模型了。

1 创建自定义模型

langchain环境，遵循标准的LLM接口，实现一个自定义LLM类主要需实现以下两个方法：

_call: 接受一个字符串（以及一些可选的停止词），返回生成的字符串。
_llm_type: 这是一个用于日志记录的属性，返回模型类型的字符串。

可选实现

_identifying_params: 返回一个用于帮助识别模型及其输出的字典。
_acall: 异步版本的_call方法。
_stream: 实现模型输出的逐token流。
_astream: 异步版本的_stream方法。

示例如下

from typing import Any, Dict, Iterator, List, Mapping, Optionalfrom langchain_core.callbacks.manager import CallbackManagerForLLMRun
from langchain_core.language_models.llms import LLM
from langchain_core.outputs import GenerationChunkclass CustomLLM(LLM):"""A custom chat model that echoes the first `n` characters of the input.When contributing an implementation to LangChain, carefully documentthe model including the initialization parameters, includean example of how to initialize the model and include any relevantlinks to the underlying models documentation or API.Example:.. code-block:: pythonmodel = CustomChatModel(n=2)result = model.invoke([HumanMessage(content="hello")])result = model.batch([[HumanMessage(content="hello")],[HumanMessage(content="world")]])"""n: int"""The number of characters from the last message of the prompt to be echoed."""def _call(self,prompt: str,stop: Optional[List[str]] = None,run_manager: Optional[CallbackManagerForLLMRun] = None,**kwargs: Any,) -> str:"""Run the LLM on the given input.Override this method to implement the LLM logic.Args:prompt: The prompt to generate from.stop: Stop words to use when generating. Model output is cut off at thefirst occurrence of any of the stop substrings.If stop tokens are not supported consider raising NotImplementedError.run_manager: Callback manager for the run.**kwargs: Arbitrary additional keyword arguments. These are usually passedto the model provider API call.Returns:The model output as a string. Actual completions SHOULD NOT include the prompt."""if stop is not None:raise ValueError("stop kwargs are not permitted.")return prompt[: self.n]def _stream(self,prompt: str,stop: Optional[List[str]] = None,run_manager: Optional[CallbackManagerForLLMRun] = None,**kwargs: Any,) -> Iterator[GenerationChunk]:"""Stream the LLM on the given prompt.This method should be overridden by subclasses that support streaming.If not implemented, the default behavior of calls to stream will be tofallback to the non-streaming version of the model and returnthe output as a single chunk.Args:prompt: The prompt to generate from.stop: Stop words to use when generating. Model output is cut off at thefirst occurrence of any of these substrings.run_manager: Callback manager for the run.**kwargs: Arbitrary additional keyword arguments. These are usually passedto the model provider API call.Returns:An iterator of GenerationChunks."""for char in prompt[: self.n]:chunk = GenerationChunk(text=char)if run_manager:run_manager.on_llm_new_token(chunk.text, chunk=chunk)yield chunk@propertydef _identifying_params(self) -> Dict[str, Any]:"""Return a dictionary of identifying parameters."""return {# The model name allows users to specify custom token counting# rules in LLM monitoring applications (e.g., in LangSmith users# can provide per token pricing for their model and monitor# costs for the given LLM.)"model_name": "CustomChatModel",}@propertydef _llm_type(self) -> str:"""Get the type of language model used by this chat model. Used for logging purposes only."""return "custom"

2 测试自定义模型

测试示例如下

llm = CustomLLM(n=5)
llm.invoke("This is a foobar thing")

await llm.ainvoke("world")

llm.batch(["woof woof woof", "meow meow meow"])

await llm.abatch(["woof woof woof", "meow meow meow"])

async for token in llm.astream("hello"):
print(token, end="|", flush=True)

3 langchain集成

自定义LLM集成到langchain的示例如下

# <!--IMPORTS:[{"imported": "ChatPromptTemplate", "source": "langchain_core.prompts", "docs": "https://python.langchain.com/api_reference/core/prompts/langchain_core.prompts.chat.ChatPromptTemplate.html", "title": "How to create a custom LLM class"}]-->
from langchain_core.prompts import ChatPromptTemplateprompt = ChatPromptTemplate.from_messages([("system", "you are a bot"), ("human", "{input}")]
)llm = CustomLLM(n=7)
chain = prompt | llmidx = 0
async for event in chain.astream_events({"input": "hello there!"}, version="v1"):print(event)idx += 1if idx > 7:# Truncatebreak

{'event': 'on_chain_start', 'run_id': 'a79babe4-fb33-44d6-b85e-55e09420ae0a', 'name': 'RunnableSequence', 'tags': [], 'metadata': {}, 'data': {'input': {'input': 'hello there!'}}, 'parent_ids': []}
{'event': 'on_prompt_start', 'name': 'ChatPromptTemplate', 'run_id': 'cbddb0e8-c0f8-4fb4-a276-d16dec173b1a', 'tags': ['seq:step:1'], 'metadata': {}, 'data': {'input': {'input': 'hello there!'}}, 'parent_ids': []}
{'event': 'on_prompt_end', 'name': 'ChatPromptTemplate', 'run_id': 'cbddb0e8-c0f8-4fb4-a276-d16dec173b1a', 'tags': ['seq:step:1'], 'metadata': {}, 'data': {'input': {'input': 'hello there!'}, 'output': ChatPromptValue(messages=[SystemMessage(content='you are a bot', additional_kwargs={}, response_metadata={}), HumanMessage(content='hello there!', additional_kwargs={}, response_metadata={})])}, 'parent_ids': []}
{'event': 'on_llm_start', 'name': 'CustomLLM', 'run_id': 'd9254fee-074b-45e7-b671-063a1abafb14', 'tags': ['seq:step:2'], 'metadata': {'ls_provider': 'custom', 'ls_model_type': 'llm'}, 'data': {'input': {'prompts': ['System: you are a bot\nHuman: hello there!']}}, 'parent_ids': []}
{'event': 'on_llm_stream', 'name': 'CustomLLM', 'run_id': 'd9254fee-074b-45e7-b671-063a1abafb14', 'tags': ['seq:step:2'], 'metadata': {'ls_provider': 'custom', 'ls_model_type': 'llm'}, 'data': {'chunk': 'S'}, 'parent_ids': []}
{'event': 'on_chain_stream', 'run_id': 'a79babe4-fb33-44d6-b85e-55e09420ae0a', 'tags': [], 'metadata': {}, 'name': 'RunnableSequence', 'data': {'chunk': 'S'}, 'parent_ids': []}
{'event': 'on_llm_stream', 'name': 'CustomLLM', 'run_id': 'd9254fee-074b-45e7-b671-063a1abafb14', 'tags': ['seq:step:2'], 'metadata': {'ls_provider': 'custom', 'ls_model_type': 'llm'}, 'data': {'chunk': 'y'}, 'parent_ids': []}
{'event': 'on_chain_stream', 'run_id': 'a79babe4-fb33-44d6-b85e-55e09420ae0a', 'tags': [], 'metadata': {}, 'name': 'RunnableSequence', 'data': {'chunk': 'y'}, 'parent_ids': []}

reference

---

如何创建自定义 LLM 类

https://www.langchain.com.cn/docs/how_to/custom_llm/

查看全文

http://www.xdnf.cn/news/17988.html