自定义解析

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

在某些情况下，你可能希望实现一个自定义解析器，将模型输出结构化为自定义格式。有两种方法可以实现自定义解析器：

使用 RunnableLambda 或 RunnableGenerator（推荐）
这是大多数情况下的推荐方法，因为它更简单且与 LangChain 的 LCEL（LangChain Expression Language）无缝集成。
继承解析器的基类
这种方法更复杂，通常不推荐，除非你有特殊需求。

以下是如何使用这两种方法实现自定义解析器的详细说明。

方法 1：使用 `RunnableLambda` 或 `RunnableGenerator`

示例 1：简单的解析器（非流式）

以下是一个简单的解析器，它将模型输出的字母大小写反转。例如，如果模型输出 "Meow"，解析器将返回 "mEOW"。

python
from langchain_anthropic.chat_models import ChatAnthropic
from langchain_core.messages import AIMessage

# 定义模型
model = ChatAnthropic(model_name="claude-2.1")

# 定义解析函数
def parse(ai_message: AIMessage) -> str:
    """解析 AI 消息并反转大小写"""
    return ai_message.content.swapcase()

# 创建链
chain = model | parse

# 调用链
result = chain.invoke("hello")
print(result)  # 输出: 'hELLO!'

注意：LCEL 会自动将 parse 函数升级为 RunnableLambda(parse)。

示例 2：流式解析器

如果你需要支持流式解析，可以使用 RunnableGenerator。以下是一个流式解析器的示例：

python
from typing import Iterable
from langchain_core.messages import AIMessageChunk
from langchain_core.runnables import RunnableGenerator

# 定义流式解析函数
def streaming_parse(chunks: Iterable[AIMessageChunk]) -> Iterable[str]:
    for chunk in chunks:
        yield chunk.content.swapcase()

# 将解析函数包装为 RunnableGenerator
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)
# 输出: i|'M| cLAUDE|,| AN| ai| ASSISTANT| CREATED| BY| aN|THROP|IC| TO| BE| HELPFUL|,| HARMLESS|,| AND| HONEST|.|

方法 2：继承解析器的基类

如果你需要更细粒度的控制，可以继承 BaseOutputParser 或 BaseGenerationOutputParser。以下是一个简单的布尔解析器示例：

示例 1：继承 `BaseOutputParser`

python
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 期望输出值为 {self.true_val} 或 {self.false_val}（不区分大小写）。"
                f"实际收到: {cleaned_text}."
            )
        return cleaned_text == self.true_val.upper()

    @property
    def _type(self) -> str:
        return "boolean_output_parser"

# 测试解析器
parser = BooleanOutputParser()
print(parser.invoke("YES"))  # 输出: True
print(parser.invoke("NO"))   # 输出: False

示例 2：继承 `BaseGenerationOutputParser`

如果你需要解析原始模型输出（例如，包含元数据的 Generation 或 ChatGeneration），可以继承 BaseGenerationOutputParser。

python
from typing import List
from langchain_core.exceptions import OutputParserException
from langchain_core.output_parsers import BaseGenerationOutputParser
from langchain_core.outputs import ChatGeneration, Generation

class StrInvertCase(BaseGenerationOutputParser[str]):
    """反转消息字符大小写的解析器"""

    def parse_result(self, result: List[Generation], *, partial: bool = False) -> str:
        if len(result) != 1:
            raise NotImplementedError("此解析器仅支持单个生成结果。")
        generation = result[0]
        if not isinstance(generation, ChatGeneration):
            raise OutputParserException("此解析器仅支持聊天生成结果。")
        return generation.message.content.swapcase()

# 测试解析器
chain = model | StrInvertCase()
print(chain.invoke("Tell me a short sentence about yourself"))
# 输出: 'hELLO! mY NAME IS cLAUDE.'

总结

推荐方法：使用 RunnableLambda 或 RunnableGenerator，因为它们更简单且与 LCEL 无缝集成。
高级方法：继承 BaseOutputParser 或 BaseGenerationOutputParser，适用于需要更细粒度控制的场景。

希望这篇教程对你有所帮助！如果你有任何问题或需要进一步的帮助，请随时提问。

目录