跳到主要内容
Open In ColabOpen on GitHub

如何创建自定义输出解析器

在某些情况下,您可能需要实现自定义 解析器,以将模型输出结构化为自定义格式。

有两种方法可以实现自定义解析器

  1. LCEL 中使用 RunnableLambdaRunnableGenerator -- 我们强烈建议在大多数用例中使用此方法
  2. 通过继承输出解析的基类之一 -- 这是执行此操作的困难方式

这两种方法之间的区别主要是表面上的,主要在于触发的回调(例如,on_chain_starton_parser_start),以及 runnable lambda 与解析器在 LangSmith 等追踪平台中的可视化方式。

Runnable Lambdas 和 Generators

推荐的解析方式是使用runnable lambdasrunnable generators

在这里,我们将创建一个简单的解析器,反转模型输出的大小写。

例如,如果模型输出:“Meow”,解析器将产生 “mEOW”。

from typing import Iterable

from langchain_anthropic.chat_models import ChatAnthropic
from langchain_core.messages import AIMessage, AIMessageChunk

model = ChatAnthropic(model_name="claude-2.1")


def parse(ai_message: AIMessage) -> str:
"""Parse the AI message."""
return ai_message.content.swapcase()


chain = model | parse
chain.invoke("hello")
'hELLO!'
提示

当使用 | 语法组合时,LCEL 会自动将函数 parse 升级为 RunnableLambda(parse)

如果您不喜欢这样,可以手动导入 RunnableLambda,然后运行 parse = RunnableLambda(parse)

流式传输是否有效?

for chunk in chain.stream("tell me about yourself in one sentence"):
print(chunk, end="|", flush=True)
i'M cLAUDE, AN ai ASSISTANT CREATED BY aNTHROPIC TO BE HELPFUL, HARMLESS, AND HONEST.|

否,它无效,因为解析器在解析输出之前会聚合输入。

如果我们想实现流式解析器,我们可以让解析器接受输入的迭代器,并在结果可用时生成结果。

from langchain_core.runnables import RunnableGenerator


def streaming_parse(chunks: Iterable[AIMessageChunk]) -> Iterable[str]:
for chunk in chunks:
yield chunk.content.swapcase()


streaming_parse = RunnableGenerator(streaming_parse)
API 参考:RunnableGenerator
重要提示

请将流式解析器包装在 RunnableGenerator 中,因为我们可能会停止使用 | 语法自动升级它。

chain = model | streaming_parse
chain.invoke("hello")
'hELLO!'

让我们确认流式传输有效!

for chunk in chain.stream("tell me about yourself in one sentence"):
print(chunk, end="|", flush=True)
i|'M| cLAUDE|,| AN| ai| ASSISTANT| CREATED| BY| aN|THROP|IC| TO| BE| HELPFUL|,| HARMLESS|,| AND| HONEST|.|

从解析基类继承

实现解析器的另一种方法是通过继承 BaseOutputParserBaseGenerationOutputParser 或其他基类解析器之一,具体取决于您需要执行的操作。

一般来说,对于大多数用例,我们建议使用这种方法,因为它会导致编写更多代码,而没有显着的好处。

最简单的输出解析器扩展了 BaseOutputParser 类,并且必须实现以下方法

  • parse:获取模型输出的字符串并进行解析
  • (可选)_type:标识解析器的名称。

当来自聊天模型或 LLM 的输出格式错误时,可以抛出 OutputParserException 以指示解析由于输入错误而失败。使用此异常允许使用解析器的代码以一致的方式处理异常。

解析器是 Runnables!🏃

由于 BaseOutputParser 实现了 Runnable 接口,因此您以这种方式创建的任何自定义解析器都将成为有效的 LangChain Runnables,并将受益于自动异步支持、批量接口、日志记录支持等。

简单解析器

这是一个简单的解析器,可以解析布尔值的字符串表示形式(例如,YESNO),并将其转换为相应的 boolean 类型。

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()

@property
def _type(self) -> str:
return "boolean_output_parser"
parser = BooleanOutputParser()
parser.invoke("YES")
True
try:
parser.invoke("MEOW")
except Exception as e:
print(f"Triggered an exception of type: {type(e)}")
Triggered an exception of type: <class 'langchain_core.exceptions.OutputParserException'>

让我们测试更改参数化

parser = BooleanOutputParser(true_val="OKAY")
parser.invoke("OKAY")
True

让我们确认其他 LCEL 方法存在

parser.batch(["OKAY", "NO"])
[True, False]
await parser.abatch(["OKAY", "NO"])
[True, False]
from langchain_anthropic.chat_models import ChatAnthropic

anthropic = ChatAnthropic(model_name="claude-2.1")
anthropic.invoke("say OKAY or NO")
API 参考:ChatAnthropic
AIMessage(content='OKAY')

让我们测试一下我们的解析器是否有效!

chain = anthropic | parser
chain.invoke("say OKAY or NO")
True
注意

解析器将适用于来自 LLM(字符串)的输出或来自聊天模型(AIMessage)的输出!

解析原始模型输出

有时,模型输出上除了原始文本之外,还有重要的附加元数据。一个例子是工具调用,其中旨在传递给被调用函数的参数在单独的属性中返回。如果您需要这种更精细的控制,您可以改为子类化 BaseGenerationOutputParser 类。

此类需要一个方法 parse_result。此方法接受原始模型输出(例如,GenerationChatGeneration 列表)并返回解析后的输出。

同时支持 GenerationChatGeneration 允许解析器与常规 LLM 以及聊天模型一起使用。

from typing import List

from langchain_core.exceptions import OutputParserException
from langchain_core.messages import AIMessage
from langchain_core.output_parsers import BaseGenerationOutputParser
from langchain_core.outputs import ChatGeneration, Generation


class 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 keep
the 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 assumed
to 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 that
partial: Whether to allow partial results. This is used for parsers
that 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 generations
raise OutputParserException(
"This output parser can only be used with a chat generation."
)
return generation.message.content.swapcase()


chain = anthropic | StrInvertCase()

让我们使用新的解析器!它应该反转模型的输出。

chain.invoke("Tell me a short sentence about yourself")
'hELLO! mY NAME IS cLAUDE.'

此页面是否对您有所帮助?