在某些情况下，您可能需要实现自定义解析器，以便将模型输出结构化为自定义格式。本文将介绍两种实现自定义解析器的方法：使用 RunnableLambda 或者 RunnableGenerator，这是我们强烈推荐的方法；以及通过继承基础类实现解析器的方法，这种方法较为复杂。
技术背景介绍
输出解析器在AI模型的实际应用中非常重要，尤其是在需要将模型的输出数据转换为符合业务需求的格式时。合理的解析器设计可以显著提高数据处理效率。
核心原理解析
-
Runnable Lambdas 和 Generators: 这种方式利用可运行的Lambda函数或者生成器函数来处理AI模型的输出。通过这种方法，我们可以直接处理模型输出，并进行实时的流解析。
-
继承解析器基础类: 在这种方式中，我们通过继承
BaseOutputParser或BaseGenerationOutputParser等基础类来实现自定义解析逻辑。尽管这种方式能提供更细粒度的控制，但通常会增加代码复杂性。
方法一: 使用 Runnable Lambda 和 Generator
这是推荐的方法，它允许快速创建轻量级的解析逻辑。我们来实现一个简单的解析器，反转模型输出的字符大小写。
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:
"""反转AI消息的字符大小写."""
return ai_message.content.swapcase()
# 使用管道运算符构建链
chain = model | parse
print(chain.invoke("hello"))
# 处理流数据的解析器
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)
# 配置流式解析链
chain = model | streaming_parse
for chunk in chain.stream("tell me about yourself in one sentence"):
print(chunk, end="|", flush=True)
方法二: 继承解析器基础类
不推荐一般用户使用的方法，但在需要自定义异常处理或复杂逻辑时可以考虑。
from langchain_core.exceptions import OutputParserException
from langchain_core.output_parsers import BaseOutputParser
# 定义布尔解析器
class BooleanOutputParser(BaseOutputParser[bool]):
"""自定义布尔解析器."""
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()
# 测试解析器
parser = BooleanOutputParser()
print(parser.invoke("YES"))
# 使用新参数测试
parser = BooleanOutputParser(true_val="OKAY")
print(parser.invoke("OKAY"))
应用场景分析
- 文本处理: 自动化转换和格式化模型输出。
- 数据清洗: 将非结构化数据转换为结构化格式。
- 复杂业务逻辑: 自定义解析器支持复杂的业务逻辑实现。
实践建议
- 对于简单解析需求，优先使用
RunnableLambda或RunnableGenerator。 - 需要复杂异常处理时，考虑继承基础解析器类。
- 提前设计解析器的输入输出格式，确保与业务需求一致。
通过以上方法，可以高效地实现自定义输出解析器，提高AI模型的应用价值。
如果遇到问题欢迎在评论区交流。
—END—