LangChain系列使用指南：组件介绍_模型输入输出_输出格式化_langchain输出json

1. 语言模型输出格式化快速入门

语言模型输出文本。但很多时候，您可能希望获得比纯文本更有结构的信息。这就是输出解析器发挥作用的地方。

输出解析器是帮助构建语言模型响应结构的类。一个输出解析器必须实现两种主要方法：

• “获取格式说明”：返回一个包含语言模型输出应如何格式化的指令字符串的方法。
• “解析”：接受一个字符串（假定为语言模型的响应）并将其解析为某种结构的方法。

还有一个可选的方法：

• “带提示解析”：接受一个字符串（假定为语言模型的响应）和一个提示（假定为生成这种响应的提示）并将其解析为某种结构。提示主要是在输出解析器希望重新尝试或修复输出的情况下提供的，需要提示信息来执行此操作。

入门指南

下面我们将介绍主要类型的输出解析器，即 PydanticOutputParser。

from langchain.output_parsers import PydanticOutputParser
from langchain.prompts import PromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field, validator
from langchain_openai import OpenAI

model = OpenAI(model_name="gpt-3.5-turbo-instruct", temperature=0.0)


# 定义您期望的数据结构。
class Joke(BaseModel):
    setup: str = Field(description="设置笑话的问题")
    punchline: str = Field(description="解答笑话的答案")

    # 您可以使用 Pydantic 轻松添加自定义验证逻辑。
    @validator("setup")
    def question_ends_with_question_mark(cls, field):
        if field[-1] != "?":
            raise ValueError("问题格式不正确！")
        return field


# 设置解析器 + 将指令注入到提示模板中。
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()},
)

# 以及一个旨在提示语言模型填充数据结构的查询。
prompt_and_model = prompt | model
output = prompt_and_model.invoke({"query": "告诉我一个笑话。"})
parser.invoke(output)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

Joke(setup='为什么鸡会过马路？', punchline='为了到达另一边！')
1

LCEL

输出解析器实现了 Runnable 接口，这是 LangChain 表达语言 (LCEL) 的基本构建块。这意味着它们支持 invoke、ainvoke、stream、astream、batch、abatch、astream_log 调用。

输出解析器接受字符串或 BaseMessage 作为输入，并可以返回任意类型。

parser.invoke(output)
1

Joke(setup='为什么鸡会过马路？', punchline='为了到达另一边！')
1

除了手动调用解析器外，我们也可以将其添加到我们的 Runnable 序列中：

chain = prompt | model | parser
chain.invoke({"query": "告诉我一个笑话。"})
1
2

Joke(setup='为什么鸡会过马路？', punchline='为了到达另一边！')
1

虽然所有解析器都支持流接口，但只有某些解析器可以通过部分解析对象进行流式处理，因为这在很大程度上取决于输出类型。不能构建部分对象的解析器将简单地生成完全解析的输出。

例如，SimpleJsonOutputParser 可以通过部分输出进行流式处理：

from langchain.output_parsers.json import SimpleJsonOutputParser

json_prompt = PromptTemplate.from_template(
    "返回一个带有 `answer` 键的 JSON 对象，回答以下问题：{question}"
)
json_parser = SimpleJsonOutputParser()
json_chain = json_prompt | model | json_parser
1
2
3
4
5
6
7

list(json_chain.stream({"question": "谁发明了显微镜？"}))
1

[{},
 {'answer': ''},
 {'answer': 'Ant'},
 {'answer': 'Anton'},
 {'answer': 'Antonie'},
 {'answer': 'Antonie van'},
 {'answer': 'Antonie van Lee'},
 {'answer': 'Antonie van Leeu'},
 {'answer': 'Antonie van Leeuwen'},
 {'answer': 'Antonie van Leeuwenho'},
 {'answer': 'Antonie van Leeuwenhoek'}]
1
2
3
4
5
6
7
8
9
10
11

而 PydanticOutputParser 则不能：

list(chain.stream({"query": "告诉我一个笑话。"}))
1

[Joke(setup='为什么鸡会过马路？', punchline='为了到达另一边！)
1

2. 自定义输出解析器

在某些情况下，您可能希望实现一个自定义解析器，将模型输出结构化为自定义格式。

有两种实现自定义解析器的方式：

1. 使用 LCEL 中的 RunnableLambda 或 RunnableGenerator – 我们强烈推荐这种方式用于大多数情况
2. 通过继承输出解析的基类之一来实现 – 这是一种较为复杂的方式

这两种方法之间的区别主要是表面的，主要体现在触发哪些回调（例如 on_chain_start vs. on_parser_start），以及可追踪平台（如 LangSmith）中可视化可运行 lambda 和解析器的方式。

可运行 Lambda 和生成器

推荐的解析方式是使用 可运行 lambda 和 可运行生成器！

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

例如，如果模型输出为：“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:
    """解析 AI 消息。"""
    return ai_message.content.swapcase()


chain = model | parse
chain.invoke("hello")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

'hELLO!'
1

[!NOTE]

当使用 | 语法组合时，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)
1
2

i'M cLAUDE, AN ai ASSISTANT CREATED BY aNTHROPIC TO BE HELPFUL, HARMLESS, AND HONEST.|
1

不，因为解析器会在解析输出之前聚合输入。

如果我们想要实现一个流式解析器，可以让解析器接受输入的可迭代对象，并在结果可用时产生结果。

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)
1
2
3
4
5
6
7
8
9

[!IMPORTANT]

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

chain = model | streaming_parse
chain.invoke("hello")
1
2

'hELLO!'
1

让我们确认流式处理是否有效！

for chunk in chain.stream("tell me about yourself in one sentence"):
    print(chunk, end="|", flush=True)
1
2

i|'M| cLAUDE|,| AN| ai| ASSISTANT| CREATED| BY| aN|THROP|IC| TO| BE| HELPFUL|,| HARMLESS|,| AND| HONEST|.|
1

继承解析基类

另一种实现解析器的方法是通过继承 BaseOutputParser、BaseGenerationOutputParser 或其他基本解析器之一，具体取决于您需要做什么。

一般来说，我们不建议大多数情况下采用这种方法，因为这样会导致编写更多代码而没有显著的好处。

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

• parse：接受模型的字符串输出并进行解析
• （可选）_type：标识解析器的名称。

当聊天模型或 LLM 的输出格式不正确时，可以抛出 OutputParserException 来指示解析失败是因为输入有误。使用此异常可以使利用解析器的代码以一致的方式处理异常。

[!TIP]

解析器是可运行的！

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小丑西瓜9/article/detail/621559