LangChain输出解析器！

LangChain 中的输出解析器

输出解析器是一种专用于处理和构建语言模型响应的类。

一个基本的输出解析器类通常需要实现两个核心方法：

get_format_instructions：

  • 这个方法需要返回一个字符串，用于指导如何格式化语言模型的输出，告诉它应该如何组织并构建它的回答。

parse：

  • 这个方法接收一个字符串（也就是语言模型的输出）并将其解析为特定的数据结构或格式。
  • 这一步通常用于确保模型的输出符合我们的预期，并且能够以我们需要的形式进行后续处理。

还有一个可选的方法。

parse_with_prompt：

这个方法接收一个字符串（也就是语言模型的输出）和一个提示（用于生成这个输出的提示），并将其解析为特定的数据结构。

这样，你可以根据原始提示来修正或重新解析模型的输出，确保输出的信息更加准确和贴合要求。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
class OutputParser:
    def __init__(self):
        pass

    def get_format_instructions(self):
        # 返回一个字符串，指导如何格式化模型的输出
        pass

    def parse(self, model_output):
        # 解析模型的输出，转换为某种数据结构或格式
        pass

    def parse_with_prompt(self, model_output, prompt):
        # 基于原始提示解析模型的输出，转换为某种数据结构或格式
        pass

LangChain通过这个三个方法，实现了各种解析器：

列表解析器（List Parser）：

这个解析器用于处理模型生成的输出，当需要模型的输出是一个列表的时候使用。

例如，如果你询问模型列出所有鲜花的库存，模型的回答应该是一个列表。

日期时间解析器（Datetime Parser）：

这个解析器用于处理日期和时间相关的输出，确保模型的输出是正确的日期或时间格式。

枚举解析器（Enum Parser）：

这个解析器用于处理预定义的一组值，当模型的输出应该是这组预定义值之一时使用。

例如，如果你定义了一个问题的答案只能是是或否，那么枚举解析器可以确保模型的回答是这两个选项之一。

结构化输出解析器（Structured Output Parser）：

这个解析器用于处理复杂的、结构化的输出。

如果你的应用需要模型生成具有特定结构的复杂回答（例如一份报告、一篇文章等），那么可以使用结构化输出解析器来实现。

Pydantic（JSON）解析器：

这个解析器用于处理模型的输出，当模型的输出应该是一个符合特定格式的JSON对象时使用。

它使用Pydantic库，这是一个数据验证库，可以用于构建复杂的数据模型，并确保模型的输出符合预期的数据模型。

自动修复解析器（Auto-Fixing Parser）：

这个解析器可以自动修复某些常见的模型输出错误。

例如，如果模型的输出应该是一段文本，但是模型返回了一段包含语法或拼写错误的文本，自动修复解析器可以自动纠正这些错误。

重试解析器（RetryWithErrorOutputParser）：

这个解析器用于在模型的初次输出不符合预期时，尝试修复或重新生成新的输出。

例如，如果模型的输出应该是一个日期，但是模型返回了一个字符串，那么重试解析器可以重新提示模型生成正确的日期格式。

Pydantic（JSON）解析器实战

创建模型实例：

1
2
3
4
5
6
7
# 设置OpenAI API密钥
import os
os.environ["OPENAI_API_KEY"] = '你的OpenAI API Key'

# 创建模型实例
from langchain import OpenAI
model = OpenAI(model_name='text-davinci-003')

定义输出数据的格式：

先创建了一个空的DataFrame，用于存储从模型生成的描述。

接下来，通过一个名为FlowerDescription的Pydantic BaseModel类，定义了期望的数据格式（也就是数据的结构）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# ------Part 2
# 创建一个空的DataFrame用于存储结果
import pandas as pd
df = pd.DataFrame(columns=["flower_type", "price", "description", "reason"])

# 数据准备
flowers = ["玫瑰", "百合", "康乃馨"]
prices = ["50", "30", "20"]

# 定义我们想要接收的数据格式
from pydantic import BaseModel, Field
class FlowerDescription(BaseModel):
    flower_type: str = Field(description="鲜花的种类")
    price: int = Field(description="鲜花的价格")
    description: str = Field(description="鲜花的描述文案")
    reason: str = Field(description="为什么要这样写这个文案")

创建输出解析器：

1
2
3
4
5
6
7
8
9
# ------Part 3
# 创建输出解析器
from langchain.output_parsers import PydanticOutputParser
output_parser = PydanticOutputParser(pydantic_object=FlowerDescription)

# 获取输出格式指示
format_instructions = output_parser.get_format_instructions()
# 打印提示
print("输出格式：",format_instructions)

让输入模型的提示和输出解析器的要求相互吻合，前后就呼应得上。

创建提示模板：

我们定义了一个提示模板，该模板将用于为模型生成输入提示。

模板中包含了你需要模型填充的变量（如价格和花的种类），以及之前获取的输出格式指示。

1
2
3
4
5
6
7
8
9
10
11
12
13
# ------Part 4
# 创建提示模板
from langchain import PromptTemplate
prompt_template = """您是一位专业的鲜花店文案撰写员。
对于售价为 {price} 元的 {flower} ，您能提供一个吸引人的简短中文描述吗？
{format_instructions}"""

# 根据模板创建提示，同时在提示中加入输出解析器的说明
prompt = PromptTemplate.from_template(prompt_template, 
       partial_variables={"format_instructions": format_instructions}) 

# 打印提示
print("提示：", prompt)

生成提示，传入模型并解析输出：

这部分是程序的主体，我们循环来处理所有的花和它们的价格。

  • 对于每种花，都根据提示模板创建了输入，然后获取模型的输出。

然后使用之前创建的解析器来解析这个输出，并将解析后的输出添加到DataFrame中。

最后，你打印出了所有的结果，并且可以选择将其保存到CSV文件中。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# ------Part 5
for flower, price in zip(flowers, prices):
    # 根据提示准备模型的输入
    input = prompt.format(flower=flower, price=price)
    # 打印提示
    print("提示：", input)

    # 获取模型的输出
    output = model(input)

    # 解析模型的输出
    parsed_output = output_parser.parse(output)
    parsed_output_dict = parsed_output.dict()  # 将Pydantic格式转换为字典

    # 将解析后的输出添加到DataFrame中
    df.loc[len(df)] = parsed_output.dict()

# 打印字典
print("输出的数据：", df.to_dict(orient='records'))

自动修复解析器（OutputFixingParser）实战

设计一个解析时出现的错误：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# 导入所需要的库和模块
from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
from typing import List

# 使用Pydantic创建一个数据格式，表示花
class Flower(BaseModel):
    name: str = Field(description="name of a flower")
    colors: List[str] = Field(description="the colors of this flower")
# 定义一个用于获取某种花的颜色列表的查询
flower_query = "Generate the charaters for a random flower."

# 定义一个格式不正确的输出
misformatted = "{'name': '康乃馨', 'colors': ['粉红色','白色','红色','紫色','黄色']}"

# 创建一个用于解析输出的Pydantic解析器，此处希望解析为Flower格式
parser = PydanticOutputParser(pydantic_object=Flower)
# 使用Pydantic解析器解析不正确的输出
parser.parse(misformatted)

这段代码如果运行，会出现错误，因为json数据格式是不正确的。

可以使用使用OutputFixingParser来帮助咱们自动解决类似的格式错误，不用人力去修改。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 从langchain库导入所需的模块
from langchain.chat_models import ChatOpenAI
from langchain.output_parsers import OutputFixingParser

# 设置OpenAI API密钥
import os
os.environ["OPENAI_API_KEY"] = '你的OpenAI API Key'

# 使用OutputFixingParser创建一个新的解析器，该解析器能够纠正格式不正确的输出
new_parser = OutputFixingParser.from_llm(parser=parser, llm=ChatOpenAI())

# 使用新的解析器解析不正确的输出
result = new_parser.parse(misformatted) # 错误被自动修正
print(result) # 打印解析后的输出结果

在于，在OutputFixingParser内部，调用了原有的PydanticOutputParser，如果成功，就返回。

如果失败，它会将格式错误的输出以及格式化的指令传递给大模型，并要求LLM进行相关的修复。

重试解析器（RetryWithErrorOutputParser）实战

OutputFixingParser不错，但它只能做简单的格式修复。

如果出错的不只是格式，比如，输出根本不完整，有缺失内容，那么仅仅根据输出和格式本身，是无法修复它的。

还是设计一个解析过程中的错误。

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
# 定义一个模板字符串，这个模板将用于生成提问
template = """Based on the user question, provide an Action and Action Input for what step should be taken.
{format_instructions}
Question: {query}
Response:"""

# 定义一个Pydantic数据格式，它描述了一个"行动"类及其属性
from pydantic import BaseModel, Field
class Action(BaseModel):
    action: str = Field(description="action to take")
    action_input: str = Field(description="input to the action")

# 使用Pydantic格式Action来初始化一个输出解析器
from langchain.output_parsers import PydanticOutputParser
parser = PydanticOutputParser(pydantic_object=Action)

# 定义一个提示模板，它将用于向模型提问
from langchain.prompts import PromptTemplate
prompt = PromptTemplate(
    template="Answer the user query.\n{format_instructions}\n{query}\n",
    input_variables=["query"],
    partial_variables={"format_instructions": parser.get_format_instructions()},
)
prompt_value = prompt.format_prompt(query="What are the colors of Orchid?")

# 定义一个错误格式的字符串
bad_response = '{"action": "search"}'
parser.parse(bad_response) # 如果直接解析，它会引发一个错误

由于bad_response只提供了action字段，而没有提供action_input字段，这与Action数据格式的预期不符，所以解析会失败。

首先尝试用OutputFixingParser来解决这个错误。

1
2
3
4
5
6
7
from langchain.output_parsers import OutputFixingParser
from langchain.chat_models import ChatOpenAI
fix_parser = OutputFixingParser.from_llm(parser=parser, llm=ChatOpenAI())
parse_result = fix_parser.parse(bad_response)
print('OutputFixingParser的parse结果:',parse_result)

OutputFixingParser的parse结果：action='search' action_input='query'

解决的问题有：

  • 不完整的数据：原始的bad_response只提供了action字段而没有action_input字段。
  • OutputFixingParser已经填补了这个缺失，为action_input字段提供了值 query。

没解决的问题有：

  • 具体性：尽管OutputFixingParser为action_input字段提供了默认值 query，但这并不具有描述性。

  • 真正的查询是 Orchid（兰花）的颜色是什么？所以，这个修复只是提供了一个通用的值，并没有真正地回答用户的问题。

可能的误导：query 可能被误解为一个指示，要求进一步查询某些内容，而不是作为实际的查询输入。

有更鲁棒的选择，最后尝试一下RetryWithErrorOutputParser这个解析器

1
2
3
4
5
6
7
8
9
10
# 初始化RetryWithErrorOutputParser，它会尝试再次提问来得到一个正确的输出
from langchain.output_parsers import RetryWithErrorOutputParser
from langchain.llms import OpenAI
retry_parser = RetryWithErrorOutputParser.from_llm(
    parser=parser, llm=OpenAI(temperature=0)
)
parse_result = retry_parser.parse_with_prompt(bad_response, prompt_value)
print('RetryWithErrorOutputParser的parse结果:',parse_result)

RetryWithErrorOutputParser的parse结果：action='search' action_input='colors of Orchid'

这个解析器成功地还原了格式，甚至也根据传入的原始提示，还原了action_input字段的内容。

RetryWithErrorOutputParser的parse结果：action=’search’，action_input=’colors of Orchid’