LangChain入门开发教程(一):Model I/O

官方文档:https://python.langchain.com/docs/get_started/introduction/

LangChain是一个能够利用大语言模型(LLM,Large Language Model)能力进行快速应用开发的框架:

  • 高度抽象的组件,可以像搭积木一样,使用LangChain的组件来实现我们的应用
  • 集成外部数据到LLM中,比如API接口数据、文件、外部应用等;
  • 提供了许多可自定义的LLM高级能力,比如Agent、RAG等等;

LangChain框架主要由以下六个部分组成:

  1. Model IO:格式化和管理LLM的输入和输出
  2. Retrieval:检索,与特定应用数据交互,比如RAG,与向量数据库密切相关,能够实现从向量数据库中搜索与问题相关的文档来作为增强LLM的上下文
  3. Agents:决定使用哪个工具(高层指令)的结构体,而tools则是允许LLM与外部系统交互的接口
  4. Chains:构建运行程序的block-style组合,即能将多个模块连接起来,实现复杂的功能应用
  5. Memory:在运行一个链路(chain)时能够存储程序状态的信息,比如存储历史对话记录,随时能够对这些历史对话记录重新加载,保证长对话的准确性
  6. Callbacks:回调机制,可以追踪任何链路的步骤,记录日志

Model IO

https://python.langchain.com/docs/modules/model_io/

Model IO

Model IO是直接与LLM交互的核心部分,从上图可以看出,它包括:

  • 对输入的格式化,涉及的组件是Prompts
  • 调用LLM进行预测,涉及的组件是LLMChat Model
  • 解析LLM的输出,涉及的组件是Output Parses

在这里先介绍下LLM和Chat Model的区别:

  • LLM可以认为是一个文本生成式模型,即大模型会根据用户输入的指示,生成一段相关的文本;
  • 而Chat Model则是一个聊天驱动的模型,即可以进行多轮对话。

Prompts

示例代码:prompt_templates.ipynb

prompt是作为用户的我们提供给LLM的一系列指示(instructions),来指导模型的生成,帮助它理解上下文,来生成相关和连贯的语言输出,比如回答问题、完成句子、进行对话等等。

prompt templates

prompt模板是一些预定义的方法,用来生成(格式化)LLM的prompt。一个模板可能包含指示、few-shot样例、适用于一个给定任务的特定的上下文和问题。

LangChain提供了一些与模型无关的模板,可以复用到不同的LLM。

PromptTemplate

  • 用于创建一个字符串prompt,其实它就是相当于的Python str.format 。它支持任意数量的变量,包括没有变量
from langchain.prompts import PromptTemplateprompt_template = PromptTemplate.from_template("Tell me a {adjective} joke about {content}."
)
prompt_template.format(adjective="funny", content="chickens")"""
'Tell me a funny joke about chickens.'
"""

ChatPromptTemplate

  • 用于创建chat model的prompt,是一个chat message的列表。

任何chat message除了相关联的文本内容,还有一个额外的参数role。比如,OpenAI的ChatGPT,一般有三种role(其他LLM基本也遵循):

  1. system,用于制作此次对话的人设,它是可选的,不是必要的。如果存在的话,则对话必须以system message为开头,但如果不存在的话,则人设就是普通的assistant,相当于使用"You are a helpful assistant."作为system message;
  2. user,作为用户的我们的输入,提供了一些诉求来让assistant(LLM)回答;
  3. assistant,一般情况是assistant(LLM)的输出,可以用来存储多轮对话(user-assistant-user-assistant-…)。但它也可以由我们来写入,用作一些样例,比如few-shot,后面再详细展开。
from langchain_core.prompts import ChatPromptTemplatechat_template = ChatPromptTemplate.from_messages([("system", "You are a helpful AI bot. Your name is {name}."),("human", "Hello, how are you doing?"),("ai", "I'm doing well, thanks!"),("human", "{user_input}"),]
)messages = chat_template.format_messages(name="Bob", user_input="What is your name?")"""
[SystemMessage(content='You are a helpful AI bot. Your name is Bob.'),HumanMessage(content='Hello, how are you doing?'),AIMessage(content="I'm doing well, thanks!"),HumanMessage(content='What is your name?')]
"""

在LangChain中,human对应上述的user,ai对应上述的assistant。

可以看到生成的prompt其实是填入format并生成对应的Message对象,因此在定义模板时也可以传入Message对象,如下:

from langchain.prompts import HumanMessagePromptTemplate
from langchain_core.messages import SystemMessagechat_template = ChatPromptTemplate.from_messages([SystemMessage(content=("You are a helpful assistant that re-writes the user's text to ""sound more upbeat.")),HumanMessagePromptTemplate.from_template("{text}"),]
)

这对于构造和复用chat prompts提供了许多灵活性。

Message Prompts

从上一个代码样例可以看出,MessagePromptTemplate 是用来构造Message对象的模板,三种role分别对应三种message模板:AIMessagePromptTemplate, SystemMessagePromptTemplate and HumanMessagePromptTemplate

而LangChain还提供另外一种可以使用任意role的message模板:ChatMessagePromptTemplate

from langchain.prompts import ChatMessagePromptTemplateprompt = "May the {subject} be with you"chat_message_prompt = ChatMessagePromptTemplate.from_template(role="Jedi", template=prompt
)
chat_message_prompt.format(subject="force")

MessagesPlaceholder

  • message占位符提供了一种对格式化时呈现什么message的完全控制,其用途在于不确定使用什么role或者想插入一个message列表的场景。
from langchain.prompts import (ChatPromptTemplate,HumanMessagePromptTemplate,MessagesPlaceholder,
)human_prompt = "Summarize our conversation so far in {word_count} words."
human_message_template = HumanMessagePromptTemplate.from_template(human_prompt)chat_prompt = ChatPromptTemplate.from_messages([MessagesPlaceholder(variable_name="conversation"), human_message_template]
)

可以看到,上述代码在chat模板中定义了一个占位符+human message的prompt,相对于确定的human message而言,message占位符则是不确定的message,可以在最终格式化呈现prompt的时候才填充数据,如下面的代码:

from langchain_core.messages import AIMessage, HumanMessagehuman_message = HumanMessage(content="What is the best way to learn programming?")
ai_message = AIMessage(content="""\
1. Choose a programming language: Decide on a programming language that you want to learn.2. Start with the basics: Familiarize yourself with the basic programming concepts such as variables, data types and control structures.3. Practice, practice, practice: The best way to learn programming is through hands-on experience\
"""
)chat_prompt.format_prompt(conversation=[human_message, ai_message], word_count="10"
).to_messages()"""
[HumanMessage(content='What is the best way to learn programming?'),AIMessage(content='1. Choose a programming language: Decide on a programming language that you want to learn.\n\n2. Start with the basics: Familiarize yourself with the basic programming concepts such as variables, data types and control structures.\n\n3. Practice, practice, practice: The best way to learn programming is through hands-on experience'),HumanMessage(content='Summarize our conversation so far in 10 words.')]
"""

Few-shot Prompt

最后再介绍一个实用性比较高,也是提示词工程经常提及的技巧:few-shot。

上面我们也提到了比如OpenAI的ChatGPT,都是包含三个不同的role:system、user、assistant。我们在user-assistant多轮对话中,塞入一些样例让大模型能够学习到这样的上下文知识。

比如这样的形式:user message_1(样例1的输入)->assistant message_1(样例1的输出,即让大模型学习应该这样回答) -> … -> user message_n(最后,放入我们真正的输入)

from langchain_core.prompts import (ChatPromptTemplate,FewShotChatMessagePromptTemplate,
)examples = [{"input": "2+2", "output": "4"},{"input": "2+3", "output": "5"},
]# This is a prompt template used to format each individual example.
example_prompt = ChatPromptTemplate.from_messages([("human", "{input}"),("ai", "{output}"),]
)
few_shot_prompt = FewShotChatMessagePromptTemplate(example_prompt=example_prompt,examples=examples,
)print(few_shot_prompt.format())
"""
Human: 2+2
AI: 4
Human: 2+3
AI: 5
"""final_prompt = ChatPromptTemplate.from_messages([("system", "You are a wondrous wizard of math."),few_shot_prompt,("human", "{input}"),]
)print(final_prompt.format(input='6+7'))
"""
System: You are a wondrous wizard of math.
Human: 2+2
AI: 4
Human: 2+3
AI: 5
Human: 6+7
"""

LCEL

在这里,先了解下LangChain的基础模块: Runnable interface,对于整个LangChain的使用会更为清晰,也就是LCEL作为LangChain的表达语言(LangChain Expression Language)。

Langchain实现了一种"Runable"协议,许多基础的组件包括上述prompt模板,以及LLM、ChatModel,以及后续的Retriever等等的调用,都继承于这个接口,当然,这也是为了更好得实现自定义的chains/组件。

它包括以下需要实现的方法:

  • invoke:使用单个输入调用chain,比如LLM
  • batch:使用一组输入调用chain
  • stream:流式返回结果

以及对应的异步方法,即需要Python的asyncio await语法实现的并发:

ainvoke、abatch、astream:分别对应上面invoke、batch、stream的异步方法

  • astream_log:回溯中间步骤的执行信息,加到最终的返回中
  • astream_event:记录流事件(events)的执行( langchain-core 0.1.14才引入的beta功能)

然后,还可以先了解下每种组件的输入和输出,方便后面的学习理解:

ComponentInput TypeOutput Type
PromptDictionaryPromptValue
ChatModelSingle string, list of chat messages or a PromptValueChatMessage
LLMSingle string, list of chat messages or a PromptValueString
OutputParserThe output of an LLM or ChatModelDepends on the parser
RetrieverSingle stringList of Documents
ToolSingle string or dictionary, depending on the toolDepends on the tool

LLMs

示例代码:llms.ipynb

Large Language Models (LLMs)是LangChain的一个核心组件,但需要注意的是,LangChain自己并不进行LLMs的服务,而是提供一个标准接口去跟许多LLMs供应商(OpenAI,Cohere等)进行交互。

具体来说,LangChain的LLMs组件是一个接收字符串作为输入,然后返回一个字符串的接口。

下面我们将使用通义千问来作为我们的LLMs供应商,目前许多国产大模型都提供了几百万的免费tokens额度,这为我们的学习用途提供了很大的便利,主打一个白嫖。

基础使用

初始化LLM.

from langchain.llms import Tongyi# api key作为参数传入
llm = Tongyi(dashscope_api_key='....')# 或者配置环境变量`DASHSCOPE_API_KEY`
llm = Tongyi()

使用LLM进行文本生成.

1、阻塞模式

llm.invoke("有什么关于失业和通货膨胀的相关性的理论")"""
'在经济学中,失业和通货膨胀之间的关系是通过一种叫做菲利普斯曲线(Phillips Curve)的理论来描述的.....'
"""

2、流式模式

for chunk in llm.stream("有什么关于失业和通货膨胀的相关性的理论"
):print(chunk, end="\n", flush=True)"""
失业
和
通
货膨胀之间的关系是
经济学中的一个重要概念,通常被描述
为菲利普斯曲线(Phill
ips Curve)的理论。该理论
由新西兰经济学家A.W.H.菲
利普斯在1958
年提出,后来被进一步发展和完善
.....
"""

3、批次调用

llm.batch(["简单介绍一个唐代诗人","简单介绍一个宋代诗人"]
)"""
['李白,字太白,号青莲居士,是唐朝时期著名的浪漫主义诗人,被誉为“诗仙”......','苏轼,字子瞻,号东坡居士,是北宋时期著名的文学家、书画家,被后人尊称为“唐宋八大家”之一......']
"""

自定义LLM

当你在本地部署了一个大模型,比如基于Ollama或者Xinference,那么你可以自定义封装LangChain的LLM接口,便可以像内置的LLMs一样使用。

首先,你需要继承LLM这个基类来自动变成上述提到的Runable,然后下面是必须重写实现的方法:

MethodDescription
_callTakes in a string and some optional stop words, and returns a string. Used by invoke.
_llm_typeA property that returns a string, used for logging purposes only.
  • 这里可以看到,LLM这个基类已经帮我们实现了invoke的方法,但它是需要调用_call方法的,返回的是字符串
  • 但其实,更多的情况下比如Tongyi等模型,是重写_generate函数来实现invoke调用,因为_generate函数返回的LLMResult结构体,可以附带更多信息,比如tokens消耗等
  • 因为整个invoke的调用顺序是这样的:BaseLLM.invoke -> LLM._generate -> LLM._call,因此当你实现了_generate函数,就不需要再去调用_call

下面几个则是非必须实现的方法:

MethodDescription
_identifying_paramsUsed to help with identifying the model and printing the LLM; should return a dictionary. This is a @property.
_acallProvides an async native implementation of _call, used by ainvoke.
_streamMethod to stream the output token by token. 看了源码,如果不实现_stream的话,则默认是调用invoke,然后转为迭代器返回
_astreamProvides an async native implementation of _stream; in newer LangChain versions, defaults to _stream.

接下来,我们还是以官网的教程和例子来实现一个非常简单的自定义LLM,它的功能仅仅是回复输入的前n个字符。

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."""n: intdef _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.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"

它的使用方法与LangChain内置LLMs是相同:

llm = CustomLLM(n=5)
print(llm)
"""
CustomLLM
Params: {'model_name': 'CustomChatModel'}
"""llm.invoke("This is a foobar thing")
"""
'This '
"""for token in llm.stream("hello"):print(token, end="|", flush=True)
"""
h|e|l|l|o|
"""llm.batch(["woof woof woof", "meow meow meow"])
"""
['woof ', 'meow ']
"""

缓存

缓存的使用场景是存在多次相同的文本生成请求,直接从缓存中获取结果进行回复,既可以提升性能,又可以减少对LLM供应商的请求,从而节省费用。

其中,最为简单的缓存方式便是内存,对应的实现类为:InMemoryCache,每次的prompt和结果都会存储在内存中。

%%timeset_llm_cache(InMemoryCache())# The first time, it is not yet in cache, so it should take longer
llm.predict("说一个笑话")
"""
CPU times: user 137 ms, sys: 77.9 ms, total: 215 ms
Wall time: 1.26 s'一朵花为什么很好笑?因为它很有梗。'
"""%%time# The second time it is, so it goes faster
llm.predict("说一个笑话")
"""
CPU times: user 752 µs, sys: 317 µs, total: 1.07 ms
Wall time: 1.03 ms'一朵花为什么很好笑?因为它很有梗。'
"""

除了内存的缓存,LangChain还提供了RedisCache这种借助外部工具的缓存实现,可以实现服务重启的缓存不失效和分布式缓存。

LangChain提供了许多内置的缓存方式:LLM Caching

消耗追踪

实现代码:callbacks、custom tongyi

LangChain内置了OpenAI API的tokens消耗追溯,可以得到一个具体请求的消耗tokens和费用,但目前只支持了OpenAI的,并且不支持stream调用的消耗追溯。我参考LangChain内置方法,实现了一套能够追踪更通用化的消耗,并且支持stream,使用方式和返回结果与内置实现完全一致。

(这里要吐槽下LangChain设计的缺陷,内置的stream调用是上层的抽象类(父类)的方法,很难把信息放在llm_output,只能放在实现类的_stream方法中返回的generation_info,但目前每一个LLMs都缺少模型名称或者tokens消耗的信息,只能自己重写对应模型的_stream方法,把信息都带入generation_info

from callbacks.manager import get_cn_llm_callbackwith get_cn_llm_callback() as cb:llm.invoke("有什么关于失业和通货膨胀的相关性的理论")print(cb)
"""
Tokens Used: 429Prompt Tokens: 18Completion Tokens: 411
Successful Requests: 1
Total Cost (CYN): ¥0.008579999999999999
"""# stream调用需要使用重写的Tongyi类
from tongyi.llm import CustomTongyi# dashscope_api_key作为参数传入
# 或者配置环境变量`DASHSCOPE_API_KEY`
llm = CustomTongyi()with get_generic_llms_callback() as cb:for chunk in llm.stream("有什么关于失业和通货膨胀的相关性的理论"):print(chunk, end="|", flush=True)print(cb)"""
在|经济学|中|,失业和通货|膨胀之间的关系被广泛研究,主要|理论有菲利普斯曲线(| Phillips Curve)和纳克斯的“|自然失业率”理论.....
Tokens Used: 301Prompt Tokens: 18Completion Tokens: 283
Successful Requests: 1
Total Cost (CYN): ¥0.006019999999999999
"""

Chat Model

示例代码:chatmodel.ipynb

正如上述提到,Chat Model则是一个聊天驱动的大语言模型,可以进行多轮对话。

LangChain内置了绝大部分市场上的大语言模型,可能部分模型没有LLM实现类,但是都会有ChatModel实现类。

Messages类型

Chat models使用chat message作为输入和输出,除了上述提到的三种基本messages类型,还有另外两种:

  1. SystemMessage:用于制作此次对话的人设;
  2. HumanMessage:用户的输入;
  3. AIMessage:模型的输出;
  4. FunctionMessage:函数调用(function call)的结果,除了对应的rolecontent参数外,还有一个name参数,表示对应名称的函数的执行结果
  5. ToolMessage:工具调用(tool call)的结果,同样有额外的参数tool_call_id,表示对应id的工具的执行结果。

函数和工具的调用暂且不在这个章节进行阐述,留到后续有专门的章节。

基础使用

其实,Chat model的使用与LLMs比较类似,只是输入和输出的格式不同。

初始化LLM.

from langchain_community.chat_models import ChatTongyi# api key作为参数传入
chat = ChatTongyi(dashscope_api_key='.....')# 或者配置环境变量`DASHSCOPE_API_KEY`
chat = ChatTongyi()

使用LLM进行文本生成.

1、阻塞模式

from langchain_core.messages import HumanMessage, SystemMessagemessages = [SystemMessage(content="你是一个数学专家"),HumanMessage(content="什么是勾股定理"),
]chat.invoke(messages)"""
AIMessage(content='勾股定理是古希腊数学家毕达哥拉斯发现的一个几何学基本定理,也被称为毕达哥拉斯定理......', response_metadata={'model_name': 'qwen-turbo', 'finish_reason': 'stop', 'request_id': '09f736aa-84b9-9a31-b249-1cf1a779b034', 'token_usage': {'input_tokens': 22, 'output_tokens': 142, 'total_tokens': 164}}, id='run-0978a38d-3611-4c89-97e2-05280f578591-0')
"""

可以看到,Chat model的invoke函数不仅返回了模型的回复内容content,还附带了response_metadata,包含一些模型调用信息。

2、流式模式

for chunk in chat.stream(messages):print(chunk.content, end="|", flush=True)"""
勾|股|定|理是古希腊数学|家毕达哥拉斯发现的一个几何|学基本定理,也被称为毕|达哥拉斯定理......
"""

当然,还有跟LLMs一样的批次调用、异步调用等等,如上面LCEL章节提到那些基础函数。

自定义Chat model

上述提到了Chat model的输入和输出都是message,因此自定义的Chat model也需要符合同样的输入和输出类型。

并且继承BaseChatModel基类,实现以下方法:

Method/PropertyDescriptionRequired/Optional
_generateUse to generate a chat result from a promptRequired
_llm_type (property)Used to uniquely identify the type of the model. Used for logging.Required
_identifying_params (property)Represent model parameterization for tracing purposes.Optional
_streamUse to implement streaming.Optional
_agenerateUse to implement a native async method.Optional
_astreamUse to implement async version of _stream.Optional

接下来,仍然是一个简单的例子,实现取prompt里最后一个message的前n个字符的功能。

from typing import Any, AsyncIterator, Dict, Iterator, List, Optionalfrom 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_executorclass CustomChatModelAdvanced(BaseChatModel):"""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")]])"""model_name: str"""The name of the model"""n: int"""The number of characters from the last message of the prompt to be echoed."""def _generate(self,messages: List[BaseMessage],stop: Optional[List[str]] = None,run_manager: 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 otherimplementation 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 itselfSHOULD BE INCLUDED as part of the output. This is not enforcedacross models right now, but it's a good practice to follow sinceit makes it much easier to parse the output of the modeldownstream and understand why generation stopped.run_manager: A run manager with callbacks for the LLM."""# Replace this with actual logic to generate a response from a list# of messages.last_message = messages[-1]tokens = last_message.content[: self.n]message = AIMessage(content=tokens,additional_kwargs={},  # Used to add additional payload (e.g., function calling request)response_metadata={  # Use for response metadata"time_in_seconds": 3,},)##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]:"""Stream the output of the model.This method should be implemented if the model can generate outputin a streaming fashion. If the model does not support streaming,do not implement it. In that case streaming requests will be automaticallyhandled 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 itselfSHOULD BE INCLUDED as part of the output. This is not enforcedacross models right now, but it's a good practice to follow sinceit makes it much easier to parse the output of the modeldownstream and understand why generation stopped.run_manager: A run manager with callbacks for the LLM."""last_message = messages[-1]tokens = last_message.content[: self.n]for token in tokens:chunk = ChatGenerationChunk(message=AIMessageChunk(content=token))if run_manager:# This is optional in newer versions of LangChain# The on_llm_new_token will be called automaticallyrun_manager.on_llm_new_token(token, chunk=chunk)yield chunk# Let's add some other information (e.g., response metadata)chunk = ChatGenerationChunk(message=AIMessageChunk(content="", response_metadata={"time_in_sec": 3}))if run_manager:# This is optional in newer versions of LangChain# The on_llm_new_token will be called automaticallyrun_manager.on_llm_new_token(token, chunk=chunk)yield chunk@propertydef _llm_type(self) -> str:"""Get the type of language model used by this chat model."""return "echoing-chat-model-advanced"@propertydef _identifying_params(self) -> Dict[str, Any]:"""Return a dictionary of identifying parameters.This information is used by the LangChain callback system, whichis used for tracing purposes make it possible to monitor LLMs."""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": self.model_name,}

与内置Chat model同样的调用方式:

model = CustomChatModelAdvanced(n=3, model_name="my_custom_model")model.invoke([HumanMessage(content="hello!"),AIMessage(content="Hi there human!"),HumanMessage(content="Meow!"),]
)"""
AIMessage(content='Meo', response_metadata={'time_in_seconds': 3}, id='run-ddb42bd6-4fdd-4bd2-8be5-e11b67d3ac29-0')
"""# 输入也支持字符串,可以等同于`[HumanMessage(content="cat vs dog")]`
for chunk in model.stream("cat vs dog"):print(chunk.content, end="|")
"""
c|a|t||
"""

缓存

缓存机制与LLMs基本也是一致的。

from langchain.globals import set_llm_cache
from langchain.cache import InMemoryCache%%timeset_llm_cache(InMemoryCache())# The first time, it is not yet in cache, so it should take longer
chat.invoke("说一个笑话")"""
CPU times: user 69.4 ms, sys: 9.48 ms, total: 78.9 ms
Wall time: 999 msAIMessage(content='小王剪了一个中分,然后他就变成了小全。', response_metadata={'model_name': 'qwen-turbo', 'finish_reason': 'stop', 'request_id': '7fb3f662-e3b1-9f2d-ae2c-0b8a1a909aec', 'token_usage': {'input_tokens': 11, 'output_tokens': 13, 'total_tokens': 24}}, id='run-8e663365-66d5-4876-9714-f88857aea39d-0')
"""%%time# The second time it is, so it goes faster
chat.invoke("说一个笑话")
"""
CPU times: user 1.27 ms, sys: 794 µs, total: 2.07 ms
Wall time: 1.33 msAIMessage(content='小王剪了一个中分,然后他就变成了小全。', response_metadata={'model_name': 'qwen-turbo', 'finish_reason': 'stop', 'request_id': '7fb3f662-e3b1-9f2d-ae2c-0b8a1a909aec', 'token_usage': {'input_tokens': 11, 'output_tokens': 13, 'total_tokens': 24}}, id='run-8e663365-66d5-4876-9714-f88857aea39d-0')
"""

同样的,除了内存的缓存,LangChain还提供了RedisCache这种借助外部工具的缓存实现,可以实现服务重启的缓存不失效和分布式缓存。

其他内置的缓存方式:LLM Caching

消耗追踪

实现代码:callbacks、custom tongyi

正如上述提到的,我参考LangChain内置方法,实现了一套能够追踪更通用化的消耗,并且支持stream,使用方式和返回结果与内置实现完全一致。 它也同样适用于Chat model,不过要在调用stream的时候记录消耗,仍然需要自己使用自定义的Chat model类。

from callbacks.manager import get_generic_llms_callbackmessages = [SystemMessage(content="你是一个数学专家"),HumanMessage(content="什么是勾股定理"),]with get_generic_llms_callback() as cb:chat.invoke(messages)print(cb)
"""
Tokens Used: 153Prompt Tokens: 22Completion Tokens: 131
Successful Requests: 1
Total Cost (CYN): ¥0.0027960000000000003
"""from tongyi.chat_model import CustomChatTongyi# dashscope_api_key作为参数传入
# 或者配置环境变量`DASHSCOPE_API_KEY`
chat = CustomChatTongyi(dashscope_api_key='....')with get_generic_llms_callback() as cb:for chunk in chat.stream(messages):print(chunk.content, end="|", flush=True)print()print(cb)
"""
勾|股|定|理是古希腊数学|家毕达哥拉斯发现的一个几何|学基本定理,也被称为毕|达哥拉斯定理......
Tokens Used: 190Prompt Tokens: 22Completion Tokens: 168
Successful Requests: 1
Total Cost (CYN): ¥0.003536
"""

模型关键参数

在这里以openai gpt为例,列举下几个影响大语言模型表现的关键参数,可以更好地理解和使用(大部分主流模型一般都是包含这些参数的):

  1. max_tokens:模型返回的最大tokens,max_tokens+input_tokens=模型支持的上下文最大长度(context length,比如gpt-3.5-turbo-16k支持最大16k tokens的长下文)
  2. temperature:采样温度,越大的值可以让模型输出更加随机
  3. top_p:同样是采样系数,表示模型会考虑前top_p概率质量的tokens
  4. frequency_penalty:频率惩罚,在生成新tokens对它们截止当前的出现频率的惩罚,越大的只可以降低模型重复相同tokens的概率
  5. presence_penalty:存在惩罚,在生成新tokens对它们截止当前已经存在的的惩罚,越大的值可以增加模型阐述新主题的概率

Output Parsers

示例代码:output_parser.ipynb

从前面可以看到,LLMChatModel的输出都是自然语言文本,也就是字符串,但在对接过程中,如果大模型能够按照特定的格式进行返回,比如常用的json,这将为开发提供极大的便捷。

结构体解析

我们可以定一个数据结构的类,主要包含一些成员变量来存储解析结果,也就是可以从大模型的输出文本中,对应这些成员变量的含义来提取对应的信息。

首先,我们定义一个关于笑话的Joke类:

  • 包含两个成员变量,description便是告诉大模型这个变量需要存储哪些信息
  • @validator("setup")可以来检验大模型解析的结果是否符合预期
from langchain.output_parsers import PydanticOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field, validator# Define your desired data structure.
class Joke(BaseModel):setup: str = Field(description="开启一个笑话的问题")punchline: str = Field(description="解答笑话的答案")# You can add custom validation logic easily with Pydantic.@validator("setup")def question_ends_with_question_mark(cls, field):if field[-1] != "?":raise ValueError("Badly formed question!")return field

接着,使用LangChain内置的解析器PydanticOutputParser,通过Prompt组件来构建提示词,输入到ChatModel组件,大模型的输出再经过解析器得到解析后的数据。

这样一个流程其实就是Chains,正如开头提到,Chains便是多个组件连接起来的序列组合,一步一步地执行,然后将结果传递给下一步。 在LangChain中使用|符号将多个组件连接起来。

# Set up a parser + inject instructions into the prompt template.
parser = PydanticOutputParser(pydantic_object=Joke)prompt = PromptTemplate(template="根据用户的输入进行解答.\n{format_instructions}\n{query}\n",input_variables=["query"],partial_variables={"format_instructions": parser.get_format_instructions()},
)# And a query intended to prompt a language model to populate the data structure.
chain = prompt | chat | parser
chain.invoke({"query": "讲一个笑话"})
"""
Joke(setup='为什么电脑永远不会感冒?', punchline='因为它有Windows(Windows,意为窗户,这里指电脑不会打开,所以不会受冷)')
"""

这里,我们进一步揭开这个解析器的面纱,先来看看解析器是如何指导提示词的:

parser.get_format_instructions()
"""
The output should be formatted as a JSON instance that conforms to the JSON schema below.
As an example, for the schema {"properties": {"foo": {"title": "Foo", "description": "a list of strings", "type": "array", "items": {"type": "string"}}}, "required": ["foo"]}
the object {"foo": ["bar", "baz"]} is a well-formatted instance of the schema. The object {"properties": {"foo": ["bar", "baz"]}} is not well-formatted.
Here is the output schema:
```
{"properties": {"setup": {"title": "Setup", "description": "开启一个笑话的问题", "type": "string"}, "punchline": {"title": "Punchline", "description": "解答笑话的答案", "type": "string"}}, "required": ["setup", "punchline"]}
```
"""

可以看到,解析器会干预提示词,指导大模型输出json格式的字符串,然后字段的含义便是我们定义的Joke类的成员变量的description,这样其实就很容易将对应的字段提取出来,写入导对应的成员变量了。

json解析

json解析器与上面的结构体解析器使用相同的提示词指导,只是最后返回dict格式的数据。

from langchain_core.output_parsers import JsonOutputParser# Set up a parser + inject instructions into the prompt template.
parser = JsonOutputParser(pydantic_object=Joke)chain = prompt | chat | parserchain.invoke({"query": "讲一个笑话"})
"""
{'setup': '为什么电脑永远不会感冒?', 'punchline': '因为它有Windows(窗户)但是不开!'}
"""

其他内置解析器

LangChain内置了许多解析器,可以去官方文档查看支持的所有解析器类型。

自定义解析器

LangChain提供了两种自定义方法:

  1. 使用RunnableLambdaRunnableGenerator,这可以符合LCEL。
  2. 继承一个输出解析器的基础类。

Runnable Lambdas.

from typing import Iterablefrom langchain_core.messages import AIMessage, AIMessageChunkdef parse(ai_message: AIMessage) -> str:"""Parse the AI message."""return ai_message.content.swapcase()chain = chat | parse
chain.invoke("hello")
"""
'hELLO! hOW CAN i ASSIST YOU TODAY?'
"""

可以看到,这种方法非常简单,仅需定义一个基本方法,接收Chat model的输出AIMessage,上面的例子的解析器是将Chat model的回复文本转换大小写。

Runnable Generators.

from langchain_core.runnables import RunnableGeneratordef streaming_parse(chunks: Iterable[AIMessageChunk]) -> Iterable[str]:for chunk in chunks:yield chunk.content.swapcase()streaming_parse = RunnableGenerator(streaming_parse)chain = chat | streaming_parsefor chunk in chain.stream("tell me about yourself in one sentence"):print(chunk, end="|", flush=True)
"""
i| AM| A| LARGE LANGUAGE MODEL CREATED BY| aLIBABA cLOUD, DESIGNED TO ANSWER QUESTIONS AND| PROVIDE INFORMATION ON VARIOUS TOPICS.|
"""

流式模式支持同样简单,仅需要定义一个方法,接收Chat model的输出AIMessageChunk迭代器,遍历对每次Chat model的流式返回进行处理,streaming_parse方法是对每一块chunk进行处理,而非上面的parse方法是对LLM返回的完整数据进行处理。

继承解析器基类.

from langchain_core.exceptions import OutputParserException
from langchain_core.output_parsers import BaseOutputParser# The [bool] desribes a parameterization of a generic.
# It's basically indicating what the return type of parse is
# in this case the return type is either True or False
class BooleanOutputParser(BaseOutputParser[bool]):"""Custom boolean parser."""true_val: str = "YES"false_val: str = "NO"def parse(self, text: str) -> bool:cleaned_text = text.strip().upper()if cleaned_text not in (self.true_val.upper(), self.false_val.upper()):raise OutputParserException(f"BooleanOutputParser expected output value to either be "f"{self.true_val} or {self.false_val} (case-insensitive). "f"Received {cleaned_text}.")return cleaned_text == self.true_val.upper()@propertydef _type(self) -> str:return "boolean_output_parser"

可以看到,其实跟上面直接定义解析方法是差不多,只不过是继承重写了parse函数。这个例子实现了:判断LLM返回是否为YES或NO,然后对应True或False的布尔值。

调用方法也是跟上面一样遵从LCEL,便不再展示了。

解析LLM原始输出.

模型的输出其实经常包含一些额外信息metadata的,因此如果解析器需要这部分信息的话,可以使用下面的方法。

from typing import Listfrom langchain_core.exceptions import OutputParserException
from langchain_core.messages import AIMessage
from langchain_core.output_parsers import BaseGenerationOutputParser
from langchain_core.outputs import ChatGeneration, Generationclass StrInvertCase(BaseGenerationOutputParser[str]):"""An example parser that inverts the case of the characters in the message.This is an example parse shown just for demonstration purposes and to keepthe example as simple as possible."""def parse_result(self, result: List[Generation], *, partial: bool = False) -> str:"""Parse a list of model Generations into a specific format.Args:result: A list of Generations to be parsed. The Generations are assumedto be different candidate outputs for a single model input.Many parsers assume that only a single generation is passed it in.We will assert for thatpartial: Whether to allow partial results. This is used for parsersthat support streaming"""if len(result) != 1:raise NotImplementedError("This output parser can only be used with a single generation.")generation = result[0]if not isinstance(generation, ChatGeneration):# Say that this one only works with chat generationsraise OutputParserException("This output parser can only be used with a chat generation.")return generation.message.content.swapcase()chain = anthropic | StrInvertCase()

因为,这里处理的 Generation是包含LLM的回复text和一些额外信息generation_info的。

并且像这个例子,还可以定义只支持Chat model的解析。

完整代码仓库

github

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.rhkb.cn/news/335845.html

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

LeetCode2336无限集中的最小数字

题目描述 现有一个包含所有正整数的集合 [1, 2, 3, 4, 5, …] 。实现 SmallestInfiniteSet 类:SmallestInfiniteSet() 初始化 SmallestInfiniteSet 对象以包含 所有 正整数。int popSmallest() 移除 并返回该无限集中的最小整数。void addBack(int num) 如果正整数 …

基于Python实现 HR 分析(逻辑回归和基于树的机器学习)【500010104】

介绍 数据集说明 此数据集包含与员工有关的综合属性集合,从人口统计细节到与工作相关的因素。该分析的主要目的是预测员工流动率并辨别导致员工流失的潜在因素。 在这个数据集中,有14,999行,10列,以及这些变量:满意度…

Python轻量级的插件框架库之pluginbase使用详解

概要 在软件开发中,插件系统是一个常见的需求。插件系统允许开发者动态加载和卸载功能模块,从而提高应用程序的灵活性和可扩展性。Python的pluginbase库是一个轻量级的插件框架,旨在简化插件系统的构建过程。pluginbase库提供了一套简单易用的API,使开发者能够快速集成插件…

leetCode-hot100-数组专题之子数组+二维数组

数组专题之子数组二维数组 子数组238.除自身以外数组的乘积560.和为K的子数组 二维数组48.旋转图像 子数组 数组的子数组问题是算法中常见的一类问题,通常涉及到数组的连续元素。在解决这类问题时,常用的方法有前缀和、滑动窗口、双指针,分治…

Sping源码(九)—— Bean的初始化(非懒加载)— getMergedLocalBeanDefinition

序言 前两篇文章介绍了Bean初始化之前的一些准备工作,包括设置BeanFacroty的ConversionService属性以及将Bean进行冻结。这篇文章将会进入到preInstantiateSingletons方法。进一步了解Bean的初始化流程。 preInstantiateSingletons public void preInstantiateSin…

一篇文章讲透排序算法之快速排序

前言 本篇博客难度较高,建议在学习过程中先阅读一遍思路、浏览一遍动图,之后研究代码,之后仔细体会思路、体会动图。之后再自己进行实现。 一.快排介绍与思想 快速排序相当于一个对冒泡排序的优化,其大体思路是先在文中选取一个…

装机必备——截图工具Snipaste安装教程

装机必备——截图工具Snipaste安装教程 软件下载 软件名称:Snipaste2.7 软件语言:简体中文 软件大小:15.37M 系统要求:Windows7或更高, 32/64位操作系统 硬件要求:CPU2GHz ,RAM2G或更高 下载通…

数据结构(七)查找

2024年5月26日一稿(王道P291) 7.1 查找的基本概念 7.2 顺序查找和折半查找 7.2.1 顺序查找 7.2.2 折半查找 7.2.3 分块查找 7.3 树形查找 7.3.1 二叉排序树(BST) 7.3.2 平衡二叉树 7.4 B树和B树 7.4.1 B树及其基本操作 7.4.2 B树的基本概念 7.5 散列&…

idea、datagrip注册记录下

一、DataGrip注册 DataGrip版本号:DataGrip 2023.2 访问地址:https://3.jetbra.in/ 点击“hardbin.com”,下载“jetbra.zip” 在vm里面添加上: -javaagent:D:\work\idea\jetbra\ja-netfilter.jarjetbrains重启datagrip 在刚刚…

【408真题】2009-28

“接”是针对题目进行必要的分析,比较简略; “化”是对题目中所涉及到的知识点进行详细解释; “发”是对此题型的解题套路总结,并结合历年真题或者典型例题进行运用。 涉及到的知识全部来源于王道各科教材(2025版&…

WebGL技术在教育培训中的应用

WebGL技术在教育培训中的应用非常广泛,通过其强大的三维图形处理能力,能够为教育培训提供更加生动、互动和沉浸式的学习体验。以下是WebGL在教育培训中的几个主要应用及其具体实现。北京木奇移动技术有限公司,专业的软件外包开发公司&#xf…

奶奶也能看懂的耦合协调度分析

不会计算?跟着文献学起来~ 案例数据连接(复制链接后粘贴到浏览器中): 耦合协调度数据​spssau.com/spssaudata.html?shareDataF363000CD033FF15E557BB75B9B0D412 假如你有这样一组数据: 如何进行计算分析耦合协调度…

蓝桥楼赛第30期-Python-第三天赛题 统计学习数据题解

楼赛 第30期 Python 模块大比拼 统计学习数据 介绍 JSON(JavaScript Object Notation, /ˈdʒeɪsən/)是一种轻量级的数据交换格式,最初是作为 JavaScript 的子集被发明的,但目前已独立于编程语言之外,成为了通用的…

淘宝API探秘:一键获取店铺所有商品的魔法之旅

在数字时代的今天,数据已经成为了商业世界中的魔法石。而对于淘宝店主或者那些想要深入探索淘宝数据的人来说,淘宝API就像是打开阿里巴巴宝藏库的钥匙。今天,我们就来一起探索如何使用淘宝API,特别是如何获取店铺所有商品的接口&a…

ElasticSearch学习篇12_《检索技术核心20讲》基础篇

背景 学习极客实践课程《检索技术核心20讲》https://time.geekbang.org/column/article/215243 课程分为基础篇、进阶篇、系统案例篇 主要记录企业课程学习过程课程大纲关键点,以文档形式记录笔记。 内容 检索技术:它是更底层的通用技术&#xff0c…

芝加哥大学最新研究:GPT-4与财务预测,重塑财务分析的未来

最近,芝加哥大学的研究团队发表了一篇突破性的研究,展示了大型语言模型(LLM),特别是 OpenAI 开发的 GPT-4,如何在财务报表分析领域取得了与专业分析师相匹配甚至超越的表现。这项研究不仅凸显了人工智能在高…

鸿蒙开发接口图形图像:【WebGL】

WebGL WebGL提供图形绘制的能力,包括对当前绘制图形的位置、颜色等进行处理。 WebGL标准图形API,对应OpenGL ES 2.0特性集。 说明: 开发前请熟悉鸿蒙开发指导文档: gitee.com/li-shizhen-skin/harmony-os/blob/master/README.md…

dp秒杀优惠券

1、全局id生成器 当用户抢购时,就会生成订单并保存到tb_voucher_order这张表中,而订单表如果使用数据库自增ID就存在一些问题: id的规律性太明显受单表数据量的限制 场景分析:如果我们的id具有太明显的规则,用户或者…

如何使用 .htaccess 删除文件扩展名

本周有一个客户,购买Hostease的虚拟主机,询问我们的在线客服,如何使用 .htaccess 删除文件扩展名?我们为用户提供相关教程,用户很快解决了遇到的问题。在此,我们分享这个操作教程,希望可以对您有…

每日复盘-20240529

20240529 六日涨幅最大: ------1--------300956--------- 英力股份 五日涨幅最大: ------1--------301361--------- 众智科技 四日涨幅最大: ------1--------301361--------- 众智科技 三日涨幅最大: ------1--------300637--------- 扬帆新材 二日涨幅最大: ------1--------30…