设为首页收藏本站
开启辅助访问

使用 LangChain 进行流式处理

您所在的位置：网站首页 › 流式输出 › 使用 LangChain 进行流式处理

使用 LangChain 进行流式处理

2024-07-11 09:19| 来源: 网络整理| 查看: 265

使用 LangChain 进行流式处理

流式处理对于基于 LLM 的应用程序对最终用户的响应至关重要。

重要的 LangChain 原语，如 LLMs、解析器、提示、检索器和代理实现了 LangChain Runnable 接口。

该接口提供了两种常见的流式内容的方法：

sync stream 和 async astream：流式处理的默认实现，从链中流式传输最终输出。async astream_events 和 async astream_log：这些方法提供了一种从链中流式传输中间步骤和最终输出的方式。

让我们看看这两种方法，并尝试了解如何使用它们。 🥷

使用 Stream​

所有Runnable对象都实现了一个名为stream的同步方法和一个名为astream的异步变体。

这些方法旨在以块的形式流式传输最终输出，只要可用就会产生每个块。

只有在程序中的所有步骤都知道如何处理输入流时，即一次处理一个输入块，并生成相应的输出块时，才能进行流式处理。

这种处理的复杂程度可以有所不同，从像发出由 LLM 生成的令牌这样的简单任务，到在整个 JSON 完成之前流式传输 JSON 结果的更具挑战性的任务。

开始探索流式处理的最佳地点是与 LLM 应用程序中最重要的组件之一 -- LLMs 本身！

LLMs 和 Chat Models​

大型语言模型及其聊天变体是基于 LLM 的应用程序的主要瓶颈。🙊

大型语言模型生成对查询的完整响应可能需要几秒钟。这远远慢于应用程序对最终用户响应感觉灵敏的~200-300 ms的阈值。

使应用程序感觉更灵敏的关键策略是显示中间进度；例如，逐个令牌从模型中流式传输输出。

# 使用人类论示例，但您可以使用您喜欢的聊天模型！from langchain.chat_models import ChatAnthropicmodel = ChatAnthropic()chunks = []async for chunk in model.astream("你好。告诉我一些关于你自己的事情"): chunks.append(chunk) print(chunk.content, end="|", flush=True) 你好|!| 我| 的名字| 是| 克劳德|。| 我| 是|一个|由|人类|创建|的|AI|助手|，|旨在|有所帮助|、|无害|和|诚实|。||

让我们检查其中一个块

chunks[0]AIMessageChunk(content=' 你好')

我们得到了一个叫做 AIMessageChunk 的东西。这个块代表了一个 AIMessage 的一部分。

消息块是可以添加的 -- 可以简单地将它们相加以获得到目前为止响应的状态！

chunks[0] + chunks[1] + chunks[2] + chunks[3] + chunks[4]AIMessageChunk(content=' 你好! 我的名字是')链​

几乎所有的 LLM 应用都涉及到不止一个调用语言模型的步骤。

让我们使用 LangChain 表达语言 (LCEL) 创建一个简单的链，它结合了一个提示、模型和一个解析器，并验证了流式处理是否有效。

我们将使用 StrOutputParser 来解析模型的输出。这是一个简单的解析器，从 AIMessageChunk 中提取 content 字段，给我们模型返回的 token。

:::{.callout-tip} LCEL 是一种通过将不同的 LangChain 原语链接在一起来指定“程序”的 声明性 方法。使用 LCEL 创建的链受益于 stream 和 astream 的自动实现，允许流式传输最终输出。事实上，使用 LCEL 创建的链实现了整个标准 Runnable 接口。 :::

from langchain_core.output_parsers import StrOutputParserfrom langchain_core.prompts import ChatPromptTemplateprompt = ChatPromptTemplate.from_template("告诉我一个关于 {topic} 的笑话")parser = StrOutputParser()chain = prompt | model | parserasync for chunk in chain.astream({"topic": "鹦鹉"}): print(chunk, end="|", flush=True) 这是|一个|关于|一只|鹦鹉|的|愚蠢|的|笑话|:|什么|样|的|老师|会|给出|好|建议|?| 一个|父母|亲(括弧)| 一个|!||

:::{.callout-note} 您不必使用 LangChain 表达语言 来使用 LangChain，您可以依赖于标准的 命令式 编程方法，通过在每个组件上分别调用 invoke、batch 或 stream，将结果分配给变量，然后根据需要在下游使用它们。

如果这符合您的需求，那么对我们来说也是可以的 👌！ :::

使用输入流​

如果您想要在生成时从输出中流式传输 JSON 怎么办？

如果您依赖于 json.loads 来解析部分 json，那么解析将失败，因为部分 json 不会是有效的 json。

您可能完全不知道该做什么

，并声称无法流式传输 JSON。

事实上，有一种方法可以做到 -- 解析器需要操作输入流，并尝试将部分 json “自动完成”为有效状态。

让我们看看这样的解析器如何运作，以理解其含义。

from langchain_core.output_parsers import JsonOutputParserchain = ( model | JsonOutputParser()) # 由于 Langchain 旧版本中的一个错误，JsonOutputParser 未能从某些模型中流式传输结果async for text in chain.astream( '以 JSON 格式输出法国、西班牙和日本的国家及其人口的列表。使用一个带有“countries”外键的字典，其中包含一个国家列表。每个国家应该有“name”和“population”关键字。'): print(text, flush=True){}{'countries': []}{'countries': [{}]}{'countries': [{'name': ''}]}{'countries': [{'name': '法国'}]}{'countries': [{'name': '法国', 'population': 67}]}{'countries': [{'name': '法国', 'population': 6739}]}{'countries': [{'name': '法国', 'population': 673915}]}{'countries': [{'name': '法国', 'population': 67391582}]}{'countries': [{'name': '法国', 'population': 67391582}, {}]}{'countries': [{'name': '法国', 'population': 67391582}, {'name': ''}]}{'countries': [{'name': '法国', 'population': 67391582}, {'name': '西'}]}{'countries': [{'name': '法国', 'population': 67391582}, {'name': '西班牙'}]}{'countries': [{'name': '法国', 'population': 67391582}, {'name': '西班牙', 'population': 46}]}{'countries': [{'name': '法国', 'population': 67391582}, {'name': '西班牙', 'population': 4675}]}{'countries': [{'name': '法国', 'population': 67391582}, {'name': '西班牙', 'population': 467547}]}{'countries': [{'name': '法国', 'population': 67391582}, {'name': '西班牙', 'population': 46754778}]}{'countries': [{'name': '法国', 'population': 67391582}, {'name': '西班牙', 'population': 46754778}, {}]}{'countries': [{'name': '法国', 'population': 67391582}, {'name': '西班牙', 'population': 46754778}, {'name': ''}]}{'countries': [{'name': '法国', 'population': 67391582}, {'name': '西班牙', 'population': 46754778}, {'name': '日本'}]}{'countries': [{'name': '法国', 'population': 67391582}, {'name': '西班牙', 'population': 46754778}, {'name': '日本', 'population': 12}]}{'countries': [{'name': '法国', 'population': 67391582}, {'name': '西班牙', 'population': 46754778}, {'name': '日本', 'population': 12647}]}{'countries': [{'name': '法国', 'population': 67391582}, {'name': '西班牙', 'population': 46754778}, {'name': '日本', 'population': 1264764}]}{'countries': [{'name': '法国', 'population': 67391582}, {'name': '西班牙', 'population': 46754778}, {'name': '日本', 'population': 126476461}]}

现在，让我们中断流式传输。我们将使用先前的示例，并在末尾附加一个提取函数，该函数从最终的 JSON 中提取国家名称。

:::{.callout-warning} 链中的任何步骤，如果操作的是最终输入而不是输入流，都可能通过 stream 或 astream 打破流式传输功能。 :::

:::{.callout-tip} 稍后，我们将讨论 astream_events API，该 API 将流式传输中间步骤的结果。即使链中包含仅操作最终输入而不是输入流的步骤，该 API 也会流式传输结果。 :::

from langchain_core.output_parsers import ( JsonOutputParser,)# 一个操作最终输入而不是输入流的函数def _extract_country_names(inputs): """一个不操作输入流并且会中断流式传输的函数。""" if not isinstance(inputs, dict): return "" if "countries" not in inputs: return "" countries = inputs["countries"] if not isinstance(countries, list): return "" country_names = [ country.get("name") for country in countries if isinstance(country, dict) ] return country_nameschain = model | JsonOutputParser() | _extract_country_namesasync for text in chain.astream( '以 JSON 格式输出法国、西班牙和日本的国家及其人口的列表。使用一个带有“countries”外键的字典，其中包含一个国家列表。每个国家应该有“name”和“population”关键字。'): print(text, end="|", flush=True)['法国', '西班牙', '日本']|生成器函数​

让我们使用可以操作输入流的生成器函数来修复流式处理。

:::{.callout-tip} 生成器函数（使用 yield 的函数）允许编写能够操作输入流的代码。 :::

from langchain_core.output_parsers import JsonOutputParserasync def _extract_country_names_streaming(input_stream): """A function that operates on input streams.""" country_names_so_far = set() async for input in input_stream: if not isinstance(input, dict): continue if "countries" not in input: continue countries = input["countries"] if not isinstance(countries, list): continue for country in countries: name = country.get("name") if not name: continue if name not in country_names_so_far: yield name country_names_so_far.add(name)chain = model | JsonOutputParser() | _extract_country_names_streamingasync for text in chain.astream( 'output a list of the countries france, spain and japan and their populations in JSON format. Use a dict with an outer key of "countries" which contains a list of countries. Each country should have the key `name` and `population`'): print(text, end="|", flush=True)France|Sp|Spain|Japan|

:::{.callout-note} 因为上面的代码依赖于 JSON 自动补全，您可能会看到部分国家名称（例如，Sp 和 Spain），这不是我们对提取结果的期望！

我们关注的是流式处理的概念，而不一定是链的结果。 :::

非流式组件​

一些内置组件，如检索器，不提供任何 streaming。如果我们尝试对它们进行stream会发生什么？ 🤨

from langchain_community.vectorstores import FAISSfrom langchain_core.output_parsers import StrOutputParserfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.runnables import RunnablePassthroughfrom langchain_openai import OpenAIEmbeddingstemplate = """Answer the question based only on the following context:{context}Question: {question}"""prompt = ChatPromptTemplate.from_template(template)vectorstore = FAISS.from_texts( ["harrison worked at kensho", "harrison likes spicy food"], embedding=OpenAIEmbeddings(),)retriever = vectorstore.as_retriever()chunks = [chunk for chunk in retriever.stream("where did harrison work?")]chunks

只有从该组件产生的最终结果被流式传输了。

这是可以接受的 🥹！ 并不是所有组件都必须实现流式传输 -- 在某些情况下，流式传输要么是不必要的、困难的，要么根本没有意义。

:::{.callout-tip} 使用非流式组件构建的 LCEL 链，在许多情况下仍然能够进行流式传输，部分输出的流式传输从链中最后一个非流式步骤之后开始。 :::

retrieval_chain = ( { "context": retriever.with_config(run_name="Docs"), "question": RunnablePassthrough(), } | prompt | model | StrOutputParser())for chunk in retrieval_chain.stream( "Where did harrison work? " "Write 3 made up sentences about this place."): print(chunk, end="|", flush=True) Based| on| the| given| context|,| the| only| information| provided| about| where| Harrison| worked| is| that| he| worked| at| Ken|sh|o|.| Since| there| are| no| other| details| provided| about| Ken|sh|o|,| I| do| not| have| enough| information| to| write| 3| additional| made| up| sentences| about| this| place|.| I| can| only| state| that| Harrison| worked| at| Ken|sh|o|.||

现在我们已经了解了 stream 和 astream 的工作原理，让我们进入流式事件的世界 🏞️。

使用流事件​

事件流是一个beta API。该 API可能会根据反馈做出一些更改。

:::{.callout-note} 引入于 langchain-core 0.1.14 版本。 :::

import langchain_corelangchain_core.__version__'0.1.18'

为了让 astream_events API 正常工作：

尽可能在代码中使用 async（例如，异步工具等）如果定义自定义函数/可运行对象，请传播回调每当使用非 LCEL 的可运行对象时，请确保在 LLM 上调用 .astream() 而不是 .ainvoke，以强制 LLM 流式传输令牌。如果有任何不符合预期的情况，请告诉我们！ :)事件参考​

下面是一个参考表，显示各种 Runnable 对象可能产生的一些事件。

:::{.callout-note} 当流式传输正确实现时，对于可运行对象来说，直到完全消耗了输入流之后才会知道输入。这意味着 inputs 通常仅包含在 end 事件中，而不是在 start 事件中。 :::

事件名称块输入输出on_chat_model_start[model name]{"messages": [[SystemMessage, HumanMessage]]}on_chat_model_stream[model name]AIMessageChunk(content="hello")on_chat_model_end[model name]{"messages": [[SystemMessage, HumanMessage]]}{"generations": [...], "llm_output": None, ...}on_llm_start[model name]{'input': 'hello'}on_llm_stream[model name]'Hello'on_llm_end[model name]'Hello human!'

| | on_chain_start | format_docs | | | | | on_chain_stream | format_docs | "hello world!, goodbye world!" | | | | on_chain_end | format_docs | | [Document(...)] | "hello world!, goodbye world!" | | on_tool_start | some_tool | | {"x": 1, "y": "2"} | | | on_tool_stream | some_tool | {"x": 1, "y": "2"} | | | | on_tool_end | some_tool | | | {"x": 1, "y": "2"} | | on_retriever_start | [retriever name] | | {"query": "hello"} | | | on_retriever_chunk | [retriever name] | {documents: [...]} | | | | on_retriever_end | [retriever name] | | {"query": "hello"} | {documents: [...]} | | on_prompt_start | [template_name] | | {"question": "hello"} | | | on_prompt_end | [template_name] | | {"question": "hello"} | ChatPromptValue(messages: [SystemMessage, ...]) |

聊天模型​

让我们首先查看聊天模型产生的事件。

events = []async for event in model.astream_events("hello", version="v1"): events.append(event)/home/eugene/src/langchain/libs/core/langchain_core/_api/beta_decorator.py:86: LangChainBetaWarning: This API is in beta and may change in the future. warn_beta(

:::{.callout-note}

嘿，API 中那个奇怪的 version="v1" 参数是什么？！ 😾

这是一个beta API，我们几乎肯定会对其进行一些更改。

该版本参数将允许我们最小化对您代码的此类更改。

简而言之，我们现在让您感到恼火，以便以后不必让您感到恼火。 :::

让我们看一看开始事件和结束事件中的一些内容。

events[:3][{'event': 'on_chat_model_start', 'run_id': '555843ed-3d24-4774-af25-fbf030d5e8c4', 'name': 'ChatAnthropic', 'tags': [], 'metadata': {}, 'data': {'input': 'hello'}}, {'event': 'on_chat_model_stream', 'run_id': '555843ed-3d24-4774-af25-fbf030d5e8c4', 'tags': [], 'metadata': {}, 'name': 'ChatAnthropic', 'data': {'chunk': AIMessageChunk(content=' Hello')}}, {'event': 'on_chat_model_stream', 'run_id': '555843ed-3d24-4774-af25-fbf030d5e8c4', 'tags': [], 'metadata': {}, 'name': 'ChatAnthropic', 'data': {'chunk': AIMessageChunk(content='!')}}]events[-2:][{'event': 'on_chat_model_stream', 'run_id': '555843ed-3d24-4774-af25-fbf030d5e8c4', 'tags': [], 'metadata': {}, 'name': 'ChatAnthropic', 'data': {'chunk': AIMessageChunk(content='')}}, {'event': 'on_chat_model_end', 'name': 'ChatAnthropic', 'run_id': '555843ed-3d24-4774-af25-fbf030d5e8c4', 'tags': [], 'metadata': {}, 'data': {'output': AIMessageChunk(content=' Hello!')}}]链​

让我们回顾一下解析流式JSON的示例链，以探索流式事件API。

chain = ( model | JsonOutputParser()) # 由于较旧版本的Langchain中存在错误，JsonOutputParser无法从某些模型中流式传输结果events = [ event async for event in chain.astream_events( 'output a list of the countries france, spain and japan and their populations in JSON format. Use a dict with an outer key of "countries" which contains a list of countries. Each country should have the key `name` and `population`', version="v1", )]

如果你检查前几个事件，你会注意到有3个不同的开始事件，而不是2个开始事件。

这三个开始事件对应于：

链（模型 + 解析器）模型解析器events[:3]

你认为如果你查看最后3个事件会看到什么？中间的事件呢？

让我们使用这个API来输出模型和解析器的流事件。我们忽略链中的开始事件、结束事件和事件。

num_events = 0async for event in chain.astream_events( 'output a list of the countries france, spain and japan and their populations in JSON format. Use a dict with an outer key of "countries" which contains a list of countries. Each country should have the key `name` and `population`', version="v1",): kind = event["event"] if kind == "on_chat_model_stream": print( f"Chat model chunk: {repr(event['data']['chunk'].content)}", flush=True, ) if kind == "on_parser_stream": print(f"Parser chunk: {event['data']['chunk']}", flush=True) num_events += 1 if num_events > 30: # 截断输出 print("...") break

由于模型和解析器都支持流式传输，我们实时看到了两个组件的流式事件！挺酷的，不是吗？🦜

过滤事件​

由于这个API产生了如此多的事件，能够对事件进行过滤是很有用的。

你可以通过组件的name、组件的tags或组件的type进行过滤。

通过名称​chain = model.with_config({"run_name": "model"}) | JsonOutputParser().with_config( {"run_name": "my_parser"})max_events = 0async for event in chain.astream_events( 'output a list of the countries france, spain and japan and their populations in JSON format. Use a dict with an outer key of "countries" which contains a list of countries. Each country should have the key `name` and `population`', version="v1", include_names=["my_parser"],): print(event) max_events += 1 if max_events > 10: # 截断输出 print("...") break通过类型​chain = model.with_config({"run_name": "model"}) | JsonOutputParser().with_config( {"run_name": "my_parser"})max_events = 0async for event in chain.astream_events( 'output a list of the countries france, spain and japan and their populations in JSON format. Use a dict with an outer key of "countries" which contains a list of countries. Each country should have the key `name` and `population`', version="v1", include_types=["chat_model"],): print(event) max_events += 1 if max_events > 10: # 截断输出 print("...") break通过标签​

:::{.callout-caution}

标签会被给定可运行组件的子组件继承。

如果你使用标签进行过滤，请确保这是你想要的。 :::

chain = (model | JsonOutputParser()).with_config({"tags": ["my_chain"]})max_events = 0async for event in chain.astream_events( 'output a list of the countries france, spain and japan and their populations in JSON format. Use a dict with an outer key of "countries" which contains a list of countries. Each country should have the key `name` and `population`', version="v1", include_tags=["my_chain"],): print(event) max_events += 1 if max_events > 10: # 截断输出 print("...") break

=======

截断输出

print("...") break

{'event': 'on_chain_start', 'run_id': '190875f3-3fb7-49ad-9b6e-f49da22f3e49', 'name': 'RunnableSequence', 'tags': ['my_chain'], 'metadata': {}, 'data': {'input': 'output a list of the countries france, spain and japan and their populations in JSON format. Use a dict with an outer key of "countries" which contains a list of countries. Each country should have the key `name` and `population`'}}{'event': 'on_chat_model_start', 'name': 'ChatAnthropic', 'run_id': 'ff58f732-b494-4ff9-852a-783d42f4455d', 'tags': ['seq:step:1', 'my_chain'], 'metadata': {}, 'data': {'input': {'messages': [[HumanMessage(content='output a list of the countries france, spain and japan and their populations in JSON format. Use a dict with an outer key of "countries" which contains a list of countries. Each country should have the key `name` and `population`')]]}}}{'event': 'on_parser_start', 'name': 'JsonOutputParser', 'run_id': '3b5e4ca1-40fe-4a02-9a19-ba2a43a6115c', 'tags': ['seq:step:2', 'my_chain'], 'metadata': {}, 'data': {}}{'event': 'on_chat_model_stream', 'name': 'ChatAnthropic', 'run_id': 'ff58f732-b494-4ff9-852a-783d42f4455d', 'tags': ['seq:step:1', 'my_chain'], 'metadata': {}, 'data': {'chunk': AIMessageChunk(content=' Here')}}{'event': 'on_chat_model_stream', 'name': 'ChatAnthropic', 'run_id': 'ff58f732-b494-4ff9-852a-783d42f4455d', 'tags': ['seq:step:1', 'my_chain'], 'metadata': {}, 'data': {'chunk': AIMessageChunk(content=' is')}}{'event': 'on_chat_model_stream', 'name': 'ChatAnthropic', 'run_id': 'ff58f732-b494-4ff9-852a-783d42f4455d', 'tags': ['seq:step:1', 'my_chain'], 'metadata': {}, 'data': {'chunk': AIMessageChunk(content=' the')}}{'event': 'on_chat_model_stream', 'name': 'ChatAnthropic', 'run_id': 'ff58f732-b494-4ff9-852a-783d42f4455d', 'tags': ['seq:step:1', 'my_chain'], 'metadata': {}, 'data': {'chunk': AIMessageChunk(content=' JSON')}}{'event': 'on_chat_model_stream', 'name': 'ChatAnthropic', 'run_id': 'ff58f732-b494-4ff9-852a-783d42f4455d', 'tags': ['seq:step:1', 'my_chain'], 'metadata': {}, 'data': {'chunk': AIMessageChunk(content=' with')}}{'event': 'on_chat_model_stream', 'name': 'ChatAnthropic', 'run_id': 'ff58f732-b494-4ff9-852a-783d42f4455d', 'tags': ['seq:step:1', 'my_chain'], 'metadata': {}, 'data': {'chunk': AIMessageChunk(content=' the')}}{'event': 'on_chat_model_stream', 'name': 'ChatAnthropic', 'run_id': 'ff58f732-b494-4ff9-852a-783d42f4455d', 'tags': ['seq:step:1', 'my_chain'], 'metadata': {}, 'data': {'chunk': AIMessageChunk(content=' requested')}}{'event': 'on_chat_model_stream', 'name': 'ChatAnthropic', 'run_id': 'ff58f732-b494-4ff9-852a-783d42f4455d', 'tags': ['seq:step:1', 'my_chain'], 'metadata': {}, 'data': {'chunk': AIMessageChunk(content=' countries')}}...### 非流式组件记住，有些组件不适合流式处理，因为它们不适用于**输入流**。虽然这些组件在使用`astream`时可能会中断最终输出的流式处理，但使用`astream_events`仍然会从支持流式处理的中间步骤中产生流式事件！```python# 不支持流式处理的函数。# 它操作的是最终的输入，而不是输入流。def _extract_country_names(inputs): """不支持流式处理的函数，会中断流式处理。""" if not isinstance(inputs, dict): return "" if "countries" not in inputs: return "" countries = inputs["countries"] if not isinstance(countries, list): return "" country_names = [ country.get("name") for country in countries if isinstance(country, dict) ] return country_nameschain = ( model | JsonOutputParser() | _extract_country_names) # 这个解析器目前只适用于OpenAI

正如预期的那样，astream API无法正常工作，因为_extract_country_names不适用于流式处理。

async for chunk in chain.astream( 'output a list of the countries france, spain and japan and their populations in JSON format. Use a dict with an outer key of "countries" which contains a list of countries. Each country should have the key `name` and `population`',): print(chunk, flush=True)

['France', 'Spain', 'Japan']

现在，让我们通过使用astream_events来确认我们仍然可以从模型和解析器中看到流式输出。

num_events = 0async for event in chain.astream_events( 'output a list of the countries france, spain and japan and their populations in JSON format. Use a dict with an outer key of "countries" which contains a list of countries. Each country should have the key `name` and `population`', version="v1",): kind = event["event"] if kind == "on_chat_model_stream": print( f"Chat model chunk: {repr(event['data']['chunk'].content)}", flush=True, ) if kind == "on_parser_stream": print(f"Parser chunk: {event['data']['chunk']}", flush=True) num_events += 1 if num_events > 30: # 截断输出 print("...") break

Chat model chunk: ' Here' Chat model chunk: ' is' Chat model chunk: ' the' Chat model chunk: ' JSON' Chat model chunk: ' with' Chat model chunk: ' the' Chat model chunk: ' requested' Chat model chunk: ' countries' Chat model chunk: ' and' Chat model chunk: ' their' Chat model chunk: ' populations' Chat model chunk: ':' Chat model chunk: '\n\n```' Chat model chunk: 'json' Parser chunk: {} Chat model chunk: '\n{' Chat model chunk: '\n ' Chat model chunk: ' "' Chat model chunk: 'countries' Chat model chunk: '":' Parser chunk: {'countries': []} Chat model chunk: ' [' Chat model chunk: '\n ' Parser chunk: {'countries': [{}]} Chat model chunk: ' {' Chat model chunk: '\n ' Chat model chunk: ' "' ...

传播回调​

:::{.callout-caution} 如果在工具中调用可运行对象，请将回调传播给可运行对象；否则，将不会生成流式事件。 :::

:::{.callout-note} 当使用RunnableLambdas或@chain装饰器时，回调会在幕后自动传播。 :::

from langchain_core.runnables import RunnableLambdafrom langchain_core.tools import tooldef reverse_word(word: str): return word[::-1]reverse_word = RunnableLambda(reverse_word)@tooldef bad_tool(word: str): """不传播回调的自定义工具。""" return reverse_word.invoke(word)async for event in bad_tool.astream_events("hello", version="v1"): print(event)

{'event': 'on_tool_start', 'run_id': 'ae7690f8-ebc9-4886-9bbe-cb336ff274f2', 'name': 'bad_tool', 'tags': [], 'metadata': {}, 'data': {'input': 'hello'}} {'event': 'on_tool_stream', 'run_id': 'ae7690f8-ebc9-4886-9bbe-cb336ff274f2', 'tags': [], 'metadata': {}, 'name': 'bad_tool', 'data': {'chunk': 'olleh'}} {'event': 'on_tool_end', 'name': 'bad_tool', 'run_id': 'ae7690f8-ebc9-4886-9bbe-cb336ff274f2', 'tags': [], 'metadata': {}, 'data': {'output': 'olleh'}}

这是一个正确传播回调的重新实现。现在你会注意到我们从reverse_word可运行对象中也得到了事件。

@tooldef correct_tool(word: str, callbacks): """正确传播回调的工具。""" return reverse_word.invoke(word, {"callbacks": callbacks})async for event in correct_tool.astream_events("hello", version="v1"): print(event)

{'event': 'on_tool_start', 'run_id': '384f1710-612e-4022-a6d4-8a7bb0cc757e', 'name': 'correct_tool', 'tags': [], 'metadata': {}, 'data': {'input': 'hello'}} {'event': 'on_chain_start', 'name': 'reverse_word', 'run_id': 'c4882303-8867-4dff-b031-7d9499b39dda', 'tags': [], 'metadata': {}, 'data': {'input': 'hello'}} {'event': 'on_chain_end', 'name': 'reverse_word', 'run_id': 'c4882303-8867-4dff-b031-7d9499b39dda', 'tags': [], 'metadata': {}, 'data': {'input': 'hello', 'output': 'olleh'}} {'event': 'on_tool_stream', 'run_id': '384f1710-612e-4022-a6d4-8a7bb0cc757e', 'tags': [], 'metadata': {}, 'name': 'correct_tool', 'data': {'chunk': 'olleh'}} {'event': 'on_tool_end', 'name': 'correct_tool', 'run_id': '384f1710-612e-4022-a6d4-8a7bb0cc757e', 'tags': [], 'metadata': {}, 'data': {'output': 'olleh'}}

如果你从Runnable Lambdas或@chains中调用可运行对象，那么回调将自动传递。

from langchain_core.runnables import RunnableLambdaasync def reverse_and_double(word: str): return await reverse_word.ainvoke(word) * 2reverse_and_double = RunnableLambda(reverse_and_double)await reverse_and_double.ainvoke("1234")async for event in reverse_and_double.astream_events("1234", version="v1"): print(event)

{'event': 'on_chain_start', 'name': 'reverse_word', 'run_id': 'c4882303-8867-4dff-b031-7d9499b39dda', 'tags': [], 'metadata': {}, 'data': {'input': '1234'}} {'event': 'on_chain_end', 'name': 'reverse_word', 'run_id': 'c4882303-8867-4dff-b031-7d9499b39dda', 'tags': [], 'metadata': {}, 'data': {'input': '1234', 'output': '4321'}} {'event': 'on_chain_start', 'name': 'reverse_word', 'run_id': 'c4882303-8867-4dff-b031-7d9499b39dda', 'tags': [], 'metadata': {}, 'data': {'input': '4321'}} {'event': 'on_chain_end', 'name': 'reverse_word', 'run_id': 'c4882303-8867-4dff-b031-7d9499b39dda', 'tags': [], 'metadata': {}, 'data': {'input': '4321', 'output': '1234'}} {'event': 'on_tool_start', 'run_id': '384f1710-612e-4022-a6d4-8a7bb0cc757e', 'name': 'reverse_and_double', 'tags': [], 'metadata': {}, 'data': {'input': '1234'}} {'event': 'on_tool_stream', 'run_id': '384f1710-612e-4022-a6d4-8a7bb0cc757e', 'tags': [], 'metadata': {}, 'name': 'reverse_and_double', 'data': {'chunk': '43214321'}} {'event': 'on_tool_end', 'name': 'reverse_and_double', 'run_id': '384f1710-612e-4022-a6d4-8a7bb0cc757e', 'tags': [], 'metadata': {}, 'data': {'output': '43214321'}}

{'event': 'on_chain_start', 'run_id': '4fe56c7b-6982-4999-a42d-79ba56151176', 'name': 'reverse_and_double', 'tags': [], 'metadata': {}, 'data': {'input': '1234'}}{'event': 'on_chain_start', 'name': 'reverse_word', 'run_id': '335fe781-8944-4464-8d2e-81f61d1f85f5', 'tags': [], 'metadata': {}, 'data': {'input': '1234'}}{'event': 'on_chain_end', 'name': 'reverse_word', 'run_id': '335fe781-8944-4464-8d2e-81f61d1f85f5', 'tags': [], 'metadata': {}, 'data': {'input': '1234', 'output': '4321'}}{'event': 'on_chain_stream', 'run_id': '4fe56c7b-6982-4999-a42d-79ba56151176', 'tags': [], 'metadata': {}, 'name': 'reverse_and_double', 'data': {'chunk': '43214321'}}{'event': 'on_chain_end', 'name': 'reverse_and_double', 'run_id': '4fe56c7b-6982-4999-a42d-79ba56151176', 'tags': [], 'metadata': {}, 'data': {'output': '43214321'}}

And with the @chain decorator:

from langchain_core.runnables import chain@chainasync def reverse_and_double(word: str): return await reverse_word.ainvoke(word) * 2await reverse_and_double.ainvoke("1234")async for event in reverse_and_double.astream_events("1234", version="v1"): print(event) {'event': 'on_chain_start', 'run_id': '7485eedb-1854-429c-a2f8-03d01452daef', 'name': 'reverse_and_double', 'tags': [], 'metadata': {}, 'data': {'input': '1234'}} {'event': 'on_chain_start', 'name': 'reverse_word', 'run_id': 'e7cddab2-9b95-4e80-abaf-4b2429117835', 'tags': [], 'metadata': {}, 'data': {'input': '1234'}} {'event': 'on_chain_end', 'name': 'reverse_word', 'run_id': 'e7cddab2-9b95-4e80-abaf-4b2429117835', 'tags': [], 'metadata': {}, 'data': {'input': '1234', 'output': '4321'}} {'event': 'on_chain_stream', 'run_id': '7485eedb-1854-429c-a2f8-03d01452daef', 'tags': [], 'metadata': {}, 'name': 'reverse_and_double', 'data': {'chunk': '43214321'}} {'event': 'on_chain_end', 'name': 'reverse_and_double', 'run_id': '7485eedb-1854-429c-a2f8-03d01452daef', 'tags': [], 'metadata': {}, 'data': {'output': '43214321'}}


【本文地址】


今日新闻

推荐新闻

CopyRight 2018-2019 办公设备维修网 版权所有 豫ICP备15022753号-3