欢迎光临
我们一直在努力

[LangChain语言模型组件的设计与实现]针对模型输出的解析(下篇)

到目前为止,我们已经介绍LangChain的语言模型和提示词模板,它们都是一个Runnable对象,而且提示词模板的输出正是模型的输入,所以两者正好可以拼接在一起组成一条LCEL链。这条链还缺一环,那就是将模型输出转换成结构化数据,以方便应用处理,我们可以借助通过BaseOutputParser类型表示的输出解析器来完成此功能。提示词模板、语言模型和输出解析器组成了模型调用“三件套”。

1. PydanticOutputParser

如果说JsonOutputParser是把LLM的输出变成一个“自由的字典”,那么PydanticOutputParser就是给这个字典加上了“严格的法律约束”。它是LangChain中最推荐用于生产环境的解析器,因为它通过Pydantic实现了数据验证、类型转换和自动文档生成。虽然我们都为它们指定了一个Pydantic模型类型,但是JsonOutputParser生成的是与之具有兼容结构的字典,但是PydanticOutputParser生成的直接就是指定类型的对象,下面的演示程序体现了这一点。

from langchain_core.output_parsers import JsonOutputParser,PydanticOutputParser
from pydantic import BaseModel, Field

class UserInfo(BaseModel):
name: str = Field(description="user's name")
age: int = Field(description="user's age")
hobbies: list[str] = Field(description="user's hobbies")
json = '{"name": "Jayden", "age": "18", "hobbies": ["hiking", "painting"]}'

parser = JsonOutputParser(pydantic_object=UserInfo)
result = parser.parse(json)
assert isinstance(result, dict)
assert result["name"] == "Jayden"
assert result["age"] == "18"
assert result["hobbies"] == ["hiking", "painting"]

parser = PydanticOutputParser(pydantic_object=UserInfo)
result = parser.parse(json)
assert isinstance(result, UserInfo)
assert result.name == "Jayden"
assert result.age == 18
assert result.hobbies == ["hiking", "painting"]

PydanticOutputParser重写的parse_result方法会调用基类的同名方法得到一个与指定Pydantic类型具有兼容结构的字典,并将其作为参数调用私有方法_parse_obj生成目标对象,针对数据成员的验证和转换就实现在这里。重写的parse方法转而调用重写的parse_result。

class PydanticOutputParser(JsonOutputParser, Generic[TBaseModel]):
pydantic_object: Annotated[type[TBaseModel], SkipValidation()]
def _parse_obj(self, obj: dict) –> TBaseModel
def parse_result(
self, result: list[Generation], *, partial: bool = False
) –> TBaseModel | None:
try:
json_object = super().parse_result(result)
return self._parse_obj(json_object)
except OutputParserException:
if partial:
return None
raise

def parse(self, text: str) –> TBaseModel:
return self.parse_result([Generation(text=text)])

def get_format_instructions(self) –> str

PydanticOutputParser重写了get_format_instructions方法,我们通过如下这个简单的程序看看两者的输出究竟有何不同。我们针对自定义的Pydantic类型Task分别创建了JsonOutputParser和PydanticOutputParser对象,并输出它们的get_format_instructions方法的返回值。

from langchain_core.output_parsers import JsonOutputParser,PydanticOutputParser
from pydantic import BaseModel, Field
from enum import Enum

class TaskStatus(int, Enum):
Pending = 0
Done = 1

class Task(BaseModel):
title: str = Field(description="Task's title")
status: TaskStatus = Field(default=TaskStatus.Pending,
description="Task's status, either pending or done")

parser = JsonOutputParser(pydantic_object=Task)
print(parser.get_format_instructions())
print("-" * 100)

parser = PydanticOutputParser(pydantic_object=Task)
print(parser.get_format_instructions())

从如下所示的输出可以看出,PydanticOutputParser生成的格式指令剔除了前面“STRICT OUTPUT FORMAT”部分的内容,这部分内容引入了一套“全防御型”提示词。

STRICT OUTPUT FORMAT:
– Return only the JSON value that conforms to the schema. Do not include any additional text, explanations, headings, or separators.
– Do not wrap the JSON in Markdown or code fences (no ```or ```json).
– Do not prepend or append any text (e.g., do not write "Here is the JSON:").
– The response must be a single top-level JSON value exactly as required by the schema (object/array/etc.), with no trailing commas or comments.

The output should be formatted as a JSON instance that conforms to the JSON schema below.

As an example, for the schema {"properties": {"foo": {"title": "Foo", "description": "a list of strings", "type": "array", "items": {"type": "string"}}}, "required": ["foo"]} the object {"foo": ["bar", "baz"]} is a well-formatted instance of the schema. The object {"properties": {"foo": ["bar", "baz"]}} is not well-formatted.

Here is the output schema (shown in a code block for readability only — do not include any backticks or Markdown in your output):

{"$defs": {"TaskStatus": {"enum": [0, 1], "title": "TaskStatus", "type": "integer"}}, "properties": {"title": {"description": "Task's title", "title": "Title", "type": "string"}, "status": {"$ref": "#/$defs/TaskStatus", "default": 0, "description": "Task's status, either pending or done"}}, "required": ["title"]}

————————————————————————————————
The output should be formatted as a JSON instance that conforms to the JSON schema below.

As an example, for the schema {"properties": {"foo": {"title": "Foo", "description": "a list of strings", "type": "array", "items": {"type": "string"}}}, "required": ["foo"]}
the object {"foo": ["bar", "baz"]} is a well-formatted instance of the schema. The object {"properties": {"foo": ["bar", "baz"]}} is not well-formatted.

Here is the output schema:

{"$defs": {"TaskStatus": {"enum": [0, 1], "title": "TaskStatus", "type": "integer"}}, "properties": {"title": {"description": "Task's title", "title": "Title", "type": "string"}, "status": {"$ref": "#/$defs/TaskStatus", "default": 0, "description": "Task's status, either pending or done"}}, "required": ["title"]}

2. XMLOutputParser

XMLOutputParser是LangChain中专门用于将 LLM 生成的XML文本转换为 Python 字典的解析器。与JsonOutputParser不同,它在安全性设计和流式增量解析上有着独特的处理方式。为了让大家了解两者之间的差异,我们改写了前面的演示程序,将LCEL链的OutputParser从JsonOutputParser替换成XMLOutputParser,其他部分保持不表。

from langchain_core.output_parsers import XMLOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

parser = XMLOutputParser(tags=["name", "age", "hobbies"])

template ="""
Extract information from the following text.
{format_instructions}
context: {query}"""

prompt = (PromptTemplate.from_template(template)
.partial(format_instructions = parser.get_format_instructions()))

llm = ChatOpenAI(
model="gpt-5.2-chat",
base_url="…",
api_key= "…"
)

chain = prompt | llm | parser

query_text = "My name is Jayden, I am 18 years old and I like hiking and painting."
print(prompt.format(query=query_text))
print("-" * 100)
result = chain.invoke({"query": query_text})
print(result)

如上面的代码片段所示,我们在创建XMLOutputParser对象时指定了代表用户三个属性的XML标签“name”、“age”和“hobbies”。最终的提示词以及调用LCEL链的结果输出如下:

Extract information from the following text.
The output should be formatted as a XML file.
1. Output should conform to the tags below.
2. If tags are not given, make them on your own.
3. Remember to always open and close all the tags.

As an example, for the tags ["foo", "bar", "baz"]:
1. String "<foo>
<bar>
<baz></baz>
</bar>
</foo>" is a well-formatted instance of the schema.
2. String "<foo>
<bar>
</foo>" is a badly-formatted instance.
3. String "<foo>
<tag>
</tag>
</foo>" is a badly-formatted instance.

Here are the output tags:

['name', 'age', 'hobbies']

context: My name is Jayden, I am 18 years old and I like hiking and painting.
————————————————————————————————
{'person': [{'name': 'Jayden'}, {'age': '18'}, {'hobbies': [{'hobby': 'hiking'}, {'hobby': 'painting'}]}]}

由于提示词包含了XMLOutputParser对象生成的指令,OpenAI API生成的内容将会以XML的格式予以响应,具体的响应内容如下:

{
"choices": [
{
"content_filter_results": {…},
"finish_reason": "stop",
"index": 0,
"logprobs": null,
"message": {
"annotations": [],
"content": "```xml\\n\\u003cperson\\u003e\\n \\u003cname\\u003eJayden\\u003c/name\\u003e\\n \\u003cage\\u003e18\\u003c/age\\u003e\\n \\u003chobbies\\u003e\\n \\u003chobby\\u003ehiking\\u003c/hobby\\u003e\\n \\u003chobby\\u003epainting\\u003c/hobby\\u003e\\n \\u003c/hobbies\\u003e\\n\\u003c/person\\u003e\\n```",
"refusal": null,
"role": "assistant"
}
}
],
"created": 1772430698,
"id": "chatcmpl-DEqTqxieILdkhjV0eAiLDwvvP93sV",
"model": "gpt-5.2-chat-2025-12-11",
"object": "chat.completion",
"prompt_filter_results": [{…}],
"system_fingerprint": null,
"usage": {… }
}

XMLOutputParser没有继承采用累加模式的BaseCumulativeTransformOutputParser类型,而是直接继承BaseTransformOutputParser类型。因为XML是树状结构,解析难度比JSON大。它内部维护了一个状态机_StreamingParser。它不是每次返回全量 XML,而是当一个完整的Tag(标签)闭合时,立即生成这个标签及其内容的字典。这实现了真正的 “流式对象产出” ,而不是 “流式文本累加” 。

class XMLOutputParser(BaseTransformOutputParser):
tags: list[str] | None = None
encoding_matcher: re.Pattern = re.compile(
r"<([^>]*encoding[^>]*)>\\n(.*)", re.MULTILINE | re.DOTALL
)
parser: Literal["defusedxml", "xml"] = "defusedxml"

def get_format_instructions(self) –> str
def parse(self, text: str) –> dict[str, str | list[Any]]
@override
def _transform(self, input: Iterator[str | BaseMessage]) –> Iterator[AddableDict]
@override
async def _atransform(
self, input: AsyncIterator[str | BaseMessage]
) –> AsyncIterator[AddableDict]

我们在创建XMLOutputParser对象需要通过初始化tags字段设置XML的标签,但该字段只是一个字符串列表,不足以描述具有层次的XML结构。如果真的具有针对深层次XML的解析需求,我们可以将XML层次的描述以自定义指令的形式写入提示词。它的encoding_matcher字段提供了一个用于提取有效内容的正则表达式,它的默认值会强行剥离 XML 的声明头(比如<?xml version="1.0" encoding="UTF-8"?>),提取出实际的标签内容。

XMLOutputParser的parser字段具有两个选项,表示采用的XML解析方案。作为默认值的“defusedxml” 表示使用defusedxml这个Python官方推荐的安全增强库,专门用于防御针对 XML 解析器的安全攻击。 由于大模型(LLM)生成的文本本质上是“不可信数据”,使用 XMLOutputParser 时必须通过这个库来确保系统安全。“xml”表示直接使用原生的xml库,它在设计时更注重性能,但是默认开启了一些可能被黑客利用的特性。

针对XML的解析实现在重写的parse和_transform/_atransform方法中。与JsonOutputParser类似,它的get_format_instructions方法输出的格式化指令不仅仅提供了单纯的XML输出的诉求,还提供了具体的格式化要求和所用的标签,并且根据标签生成了实例示范。

3. ListOutputParser

在一段“自由的文本”中,字符串具有一般具有几种典型的表达方式,比如使用逗号分隔符、分行并使用数字需要前缀以及Markdown中使用“-”前缀。为了将这种形式的内容统一解析成字符串列表,LangChain定义了ListOutputParser,它的子类CommaSeparatedListOutputParser、NumberedListOutputParser和MarkdownListOutputParser针对的就是上述的三种文本格式。下面这个简单的程序演示了针对这三种列表输出解析器的使用。

from langchain_core.output_parsers import (
CommaSeparatedListOutputParser,
NumberedListOutputParser,
MarkdownListOutputParser
)
result = ["foo", "bar", "baz"]
parser = CommaSeparatedListOutputParser()
assert parser.parse("foo, bar, baz") == result

parser = NumberedListOutputParser()
assert parser.parse("""
1. foo
2. bar
3. baz
"""
) == result

parser = MarkdownListOutputParser()
assert parser.parse("""
– foo
– bar
– baz
"""
) == result

基类ListOutputParser定义如下,它继承自BaseTransformOutputParser。针对字符串列表的解析是利用重写的_transform方法实现的,但是具体的实现利用抽象的parse方法和另一个没有实现的parse_iter下放给继承它的子类。这两个方法体现了两种不同的实现方法,parse使用“蛮力”解析原始文本,而parse_iter使用高效的正则表达式的方式。

class ListOutputParser(BaseTransformOutputParser[list[str]]):
@abstractmethod
def parse(self, text: str) –> list[str]
def parse_iter(self, text: str) –> Iterator[re.Match]:
raise NotImplementedError

@override
def _transform(self, input: Iterator[str | BaseMessage]) –> Iterator[list[str]]:
buffer = ""
for chunk in input:
if isinstance(chunk, BaseMessage):
# Extract text
chunk_content = chunk.content
if not isinstance(chunk_content, str):
continue
buffer += chunk_content
else:
# Add current chunk to buffer
buffer += chunk

try:
done_idx = 0
# Yield only complete parts
for m in droplastn(self.parse_iter(buffer), 1):
done_idx = m.end()
yield [m.group(1)]
buffer = buffer[done_idx:]
except NotImplementedError:
parts = self.parse(buffer)
# Yield only complete parts
if len(parts) > 1:
for part in parts[:–1]:
yield [part]
buffer = parts[–1]
# Yield the last part
for part in self.parse(buffer):
yield [part]

@override
async def _atransform(
self, input: AsyncIterator[str | BaseMessage]
) –> AsyncIterator[list[str]]

对于流式输出,字符会逐个进来,所以_transform采用缓冲机制将所有字符拼接起来统一处理。在流式解析中,解析器永远不知道当前的最后一项是否已经写完。例如,模型输出“1. apple\\n2. Ban”,解析器能确定“apple”输出完成,但不能确定“ban”是“banana”还是“bandage”。_transform采用去除最后一项,并将其留待下次解析。拼接完成后,它首先调用parse_iter方法采用高效的正则表达式进行解析,如果此方法没有实现,在使用parse方法兜底。_atransform也做了类似的重写。

CommaSeparatedListOutputParser、NumberedListOutputParser和MarkdownListOutputParser通过实现的parse方法各自完成了源自不同格式的解析任务,后两个类型还重写了parse_iter方法提供了针对正则表达式的解决方案。除此之外,它们都通过重写get_format_instructions方法提供相应的格式指令指导模型按照它们希望的格式输出内容。

class CommaSeparatedListOutputParser(ListOutputParser):
@override
def get_format_instructions(self) –> str:
return (
"Your response should be a list of comma separated values, "
"eg: `foo, bar, baz` or `foo,bar,baz`"
)

@override
def parse(self, text: str) –> list[str]:
try:
reader = csv.reader(
StringIO(text), quotechar='"', delimiter=",", skipinitialspace=True
)
return [item for sublist in reader for item in sublist]
except csv.Error:
# Keep old logic for backup
return [part.strip() for part in text.split(",")]

class NumberedListOutputParser(ListOutputParser):
pattern: str = r"\\d+\\.\\s([^\\n]+)"
@override
def get_format_instructions(self) –> str:
return (
"Your response should be a numbered list with each item on a new line. "
"For example: \\n\\n1. foo\\n\\n2. bar\\n\\n3. baz"
)

def parse(self, text: str) –> list[str]:
return re.findall(self.pattern, text)

@override
def parse_iter(self, text: str) –> Iterator[re.Match]:
return re.finditer(self.pattern, text)

class MarkdownListOutputParser(ListOutputParser):
pattern: str = r"^\\s*[-*]\\s([^\\n]+)$"

@override
def get_format_instructions(self) –> str:
return "Your response should be a markdown list, eg: `- foo\\n- bar\\n- baz`"

def parse(self, text: str) –> list[str]:
return re.findall(self.pattern, text, re.MULTILINE)

@override
def parse_iter(self, text: str) –> Iterator[re.Match]:
return re.finditer(self.pattern, text, re.MULTILINE)

本节内容从作为基类的BaseLLMOutputParser和BaseOutputParser将其,介绍了一系列针对不同源文本格式和目标类型的输出解析器类型,这些类型之间的关系体现在如下所示的UML类图中。

Alternative Text

赞(0)
未经允许不得转载:171主机测评 » [LangChain语言模型组件的设计与实现]针对模型输出的解析(下篇)
分享到: 更多 (0)

评论 抢沙发

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址