到目前为止，我们已经介绍LangChain的语言模型和提示词模板，它们都是一个Runnable对象，而且提示词模板的输出正是模型的输入，所以两者正好可以拼接在一起组成一条LCEL链。这条链还缺一环，那就是将模型输出转换成结构化数据，以方便应用处理，我们可以借助通过BaseOutputParser类型表示的输出解析器来完成此功能。提示词模板、语言模型和输出解析器组成了模型调用“三件套”。
1. BaseOutputParser
常用的输出解析器类型以抽象类BaseOutputParser为基类，而该类又继承自如下这个BaseLLMOutputParser类型。这是一个泛型抽象类，泛型参数T表示解析生成的目标类型。从抽象方法parse_result的定义可以看出，被解析的输入是一个Generation列表。通过前面针对语言模型的介绍我们知道，作为语言模型基类的BaseLanguageModel的核心方法generate/agenerate会返回一个LLMResult对象，它的核心字段generations返回的就是一个两层Generation列表。这进步证实了BaseLLMOutputParser解析的正是语言模型的输出。parse_result/aparse_result方法的partial参数是为支持 “流式输出” 设计的。如果被设置为True，意味着当前的提供的内容可能只是完整回复的一部分。
class BaseLLMOutputParser(ABC, Generic[T]):
@abstractmethod
def parse_result(self, result: list[Generation], *, partial: bool = False) –> T:
async def aparse_result(
self, result: list[Generation], *, partial: bool = False
) –> T:
return await run_in_executor(None, self.parse_result, result, partial=partial)
BaseOutputParser同时继承BaseLLMOutputParser和RunnableSerializable。作为一个Runnable对象，它的输入类型LanguageModelOutput是针对BaseMessage和字符串类型的联合，这两种类型分别代表Completion模型和Chat模型的输出类型。
LanguageModelOutput = BaseMessage | str
class BaseOutputParser(
BaseLLMOutputParser, RunnableSerializable[LanguageModelOutput, T]
):
@override
def invoke(
self,
input: str | BaseMessage,
config: RunnableConfig | None = None,
**kwargs: Any,
) –> T:
if isinstance(input, BaseMessage):
return self._call_with_config(
lambda inner_input: self.parse_result(
[ChatGeneration(message=inner_input)]
),
input,
config,
run_type="parser",
)
return self._call_with_config(
lambda inner_input: self.parse_result([Generation(text=inner_input)]),
input,
config,
run_type="parser",
)
@override
def parse_result(self, result: list[Generation], *, partial: bool = False) –> T:
return self.parse(result[0].text)
@abstractmethod
def parse(self, text: str) –> T
@override
async def ainvoke(
self,
input: str | BaseMessage,
config: RunnableConfig | None = None,
**kwargs: Any | None,
) –> T
async def aparse_result(
self, result: list[Generation], *, partial: bool = False
) –> T
async def aparse(self, text: str) –> T
实现的parse_result方法会提取输入列表中的第一个Generation对象的文本内容，并将其作为输入调用抽象方法parse得到目标对象。invoke方法最终调用的是parse_result方法，对于Chat模型输出的消息或者Completion模型输出的字符串文本，该方法分别创建对应的ChatGeneration和Generation列表作为调用parse_result方法的输入参数。BaseOutputParser同时也定义了aparse方法，默认实现通过调用parse方法完成。重写的aparse_result方法会调用aparse方法，而重写的ainvoke方法又会调用aparse_result方法。
输出解析不仅仅是被动地处理语言模型生成的内容，它会根据输出的结构生成相应的指令（作为提示词）指导模型生成期望的格式（比如CSV和JSON），这一个功能体现在get_format_instructions方法上。由于这不是必需的操作，所以它并没有被定义成抽象方法，支持“格式指令生成”的输出解析器类型需要抽血此方法。它也定义了dict方法通过生成的字典解决序列化问题。
class BaseOutputParser
def get_format_instructions(self) –> str:
raise NotImplementedError
def dict(self, **kwargs: Any) –> dict
2. BaseTransformOutputParser
BaseTransformOutputParser是LangChain解析器体系中针对 “流式处理（Streaming）” 的基类。如果说BaseLLMOutputParser关注的是一次性解析完整结果，那么BaseTransformOutputParser关注的就是 “边生成边解析” 。它针对流出输出的支持是通过重写transform/atransform方法实现的，而两个方法会调用定义的私有方法_transform/_atransform，后者最终调用的依旧是parse_result/aparse_result方法。
class BaseTransformOutputParser(BaseOutputParser[T]):
def _transform(
self,
input: Iterator[str | BaseMessage],
) –> Iterator[T]:
for chunk in input:
if isinstance(chunk, BaseMessage):
yield self.parse_result([ChatGeneration(message=chunk)])
else:
yield self.parse_result([Generation(text=chunk)])
async def _atransform(
self,
input: AsyncIterator[str | BaseMessage],
) –> AsyncIterator[T]:
async for chunk in input:
if isinstance(chunk, BaseMessage):
yield await run_in_executor(
None, self.parse_result, [ChatGeneration(message=chunk)]
)
else:
yield await run_in_executor(
None, self.parse_result, [Generation(text=chunk)]
)
@override
def transform(
self,
input: Iterator[str | BaseMessage],
config: RunnableConfig | None = None,
**kwargs: Any,
) –> Iterator[T]:
yield from self._transform_stream_with_config(
input, self._transform, config, run_type="parser"
)
@override
async def atransform(
self,
input: AsyncIterator[str | BaseMessage],
config: RunnableConfig | None = None,
**kwargs: Any,
) –> AsyncIterator[T]:
async for chunk in self._atransform_stream_with_config(
input, self._atransform, config, run_type="parser"
):
yield chunk
3. BaseCumulativeTransformOutputParser
BaseCumulativeTransformOutputParser相比于BaseTransformOutputParser，其核心差异在于 “状态累加” 与 “增量控制” 。它解析的是截止到当前时刻的完整上下文，而BaseTransformOutputParser解析的仅仅是当前的片段。所谓的累加处理体现在它会将新 片段与之前的所有片段合并后再解析，这一点可以从它重写的_transform方法中看出来，_atransform方法也采用类似的方式进行了重写。
class BaseCumulativeTransformOutputParser(BaseTransformOutputParser[T]):
diff: bool = False
def _diff(
self,
prev: T | None,
next: T,
) –> T:
raise NotImplementedError
@override
def _transform(self, input: Iterator[str | BaseMessage]) –> Iterator[Any]:
prev_parsed = None
acc_gen: GenerationChunk | ChatGenerationChunk | None = None
for chunk in input:
chunk_gen: GenerationChunk | ChatGenerationChunk
if isinstance(chunk, BaseMessageChunk):
chunk_gen = ChatGenerationChunk(message=chunk)
elif isinstance(chunk, BaseMessage):
chunk_gen = ChatGenerationChunk(
message=BaseMessageChunk(**chunk.model_dump())
)
else:
chunk_gen = GenerationChunk(text=chunk)
acc_gen = chunk_gen if acc_gen is None else acc_gen + chunk_gen
parsed = self.parse_result([acc_gen], partial=True)
if parsed is not None and parsed != prev_parsed:
if self.diff:
yield self._diff(prev_parsed, parsed)
else:
yield parsed
prev_parsed = parsed
@override
async def _atransform(
self, input: AsyncIterator[str | BaseMessage]
) –> AsyncIterator[T]
除了累加机制，BaseCumulativeTransformOutputParser还支持差分模式，如果利用diff字段开启了差分模式，它会调用_diff方法，计算当前结果与上一次结果的“差额”，只返回新增的部分。这在某些流式前端更新中非常有用，支持此模式的子类需要重写_diff方法。
如下的程序演示了BaseTransformOutputParser和BaseCumulativeTransformOutputParser之间的差别。我们分别继承这两个类型定义了StrOutputParser和StrCumulativeOutputParser，实现的parse方法直接返回带解析的文本。我们将同一个字符串列表的迭代器作为参数调用它们的transform方法，并将生成的字符串组合成一个列表。从断言可以看出“累积效应”出现在StrCumulativeOutputParser上。
from langchain_core.output_parsers import （BaseTransformOutputParser,
BaseCumulativeTransformOutputParser）
class StrOutputParser(BaseTransformOutputParser[str]):
def parse(self, text: str) –> str:
return text
class StrCumulativeOutputParser(BaseCumulativeTransformOutputParser[str]):
def parse(self, text: str) –> str:
return text
result = [item for item in StrOutputParser().transform(iter(["foo", "bar","baz"]))]
assert result == ["foo", "bar","baz"]
result = [item for item in StrCumulativeOutputParser().transform(iter(["foo", "bar","baz"]))]
assert result == ["foo", "foobar","foobarbaz"]
在默写场景下应用差分模式会很有用。比如它可以减少带宽，在前端UI渲染时，如果解析器返回的是整个JSON（比如有 100 个字段），每多一个字符就传一遍全量数据非常浪费。使用diff模式，前端只接收变更指令。还可以根据 diff 产出的特定元素触发特定的逻辑。例如当列表中新出现 “报警” 这个词时，立即发送通知，而不需要每次都扫描整个列表。下面的演示程序定义了一个继承自BaseCumulativeTransformOutputParser的ListCumulativeOutputParser，它将以逗号分隔的文本内容解析为列表。我们通过重写的_diff方法剔除重复的元素。
from langchain_core.output_parsers import BaseCumulativeTransformOutputParser
class ListCumulativeOutputParser(BaseCumulativeTransformOutputParser[list[str]]):
def _diff(self, prev: list[str] | None, next: list[str]) –> list[str]:
return [item for item in next if item not in prev] if prev else next
def parse(self, text: str) –> list[str]:
return text.strip(",").split(",")
result = [item for item in ListCumulativeOutputParser(diff=True).transform(iter(["foo,bar", ",bar,baz,", "qux"]))]
assert result == [['foo', 'bar'], ['baz'], ['qux']]
result = [item for item in ListCumulativeOutputParser(diff=False).transform(iter(["foo,bar", ",bar,baz,", "qux"]))]
assert result ==[['foo', 'bar'], ['foo', 'bar', 'bar', 'baz'], ['foo', 'bar', 'bar', 'baz', 'qux']]
4. StrOutputParser
在演示BaseTransformOutputParser和BaseCumulativeTransformOutputParser差异的程序中，我们定义了一个将字符串作为解析目标类型的StrOutputParser，其实这样的类型也是LangChain众多预定义输出解析类型的一员，而且它采用与我们一致的定义方式。
class StrOutputParser(BaseTransformOutputParser[str]):
@override
def parse(self, text: str) –> str:
"""Returns the input text with no changes."""
return text
5. JsonOutputParser
继承自BaseCumulativeTransformOutputParser 的JsonOutputParser是 LangChain 中最实用、设计最精巧的解析器之一，SimpleJsonOutputParser是它的一个别名。它结合了结构化指令注入、容错解析和实时流式反馈。对于不使用函数调用来获取结构化数据的情况，这可能是最可靠的解析器。它通过纯文本提示词引导模型输出 JSON，并处理模型可能夹带的Markdown代码块（如```json ）。
因为它继承了BaseCumulativeTransformOutputParser，随着模型吐词，它会不断产出当前已完整解析出的JSON对象。如果通过设置了diff=True开启了差分模式，它会利用jsonpatch库计算并只返回新老JSON之间的变化差异，极大地节省了前端处理增量数据时的开销。它还支持部分解析，在 如果参数partial=True 并且当 JSON 还没写完（比如括号没闭合）时，parse_result会捕获JSONDecodeError并静默返回None或已识别的部分，而不是崩溃。
使用JsonOutputParser最优雅的方式是结合Pydantic。这能让LangChain自动为你生成Prompt指令，并确保返回的是强类型的Python对象。以下是一个完整的代码示例，演示如何从一段杂乱的文本中提取结构化的用户信息。我们创建了一个JsonOutputParser对象，并将其pydantic_object字段设置为我们自定义的Pydantic模型类型UserInfo。然后我们创建了一个PromptTemplate对象，生成的提示词旨在告诉模型从指定的一段文字描述中提取出个人信息。模板中除了定义表示查询输入的变量“query”外，还具有一个表示“格式化指令”的变量“format_instructions”,我们调用partial将此变量的值设置为当前JsonOutputParser对象的get_format_instructions方法返回值。
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
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")
parser = JsonOutputParser(pydantic_object=UserInfo)
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)
在创建了作为模型的ChatOpenAI对象后，我们将“三件套”组成一个LCEL链。我们提供一段包含个人性息的文本作为参数调用此链，并输出返回的结果。在调用之前，我们使用相同的输入对提示词模板进行格式化，并输出了格式化提示词文本。如下所示的是程序完整的输出，可以看出JsonOutputParser的get_format_instructions方法提供了一个非常完整而丰富的指令文本，最终得到的JSON也确实于UserInfo具有匹配的Schema。
Extract information from the following text.
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):
{“properties”: {“name”: {“description”: “user’s name”, “title”: “Name”, “type”: “string”}, “age”: {“description”: “user’s age”, “title”: “Age”, “type”: “integer”}, “hobbies”: {“description”: “user’s hobbies”, “items”: {“type”: “string”}, “title”: “Hobbies”, “type”: “array”}}, “required”: [“name”, “age”, “hobbies”]}
context: My name is Jayden, I am 18 years old and I like hiking and painting.
————————————————————————————————
{'name': 'Jayden', 'age': 18, 'hobbies': ['hiking', 'painting']}
如果我们拦截针对OpenAI API调用的HTTP请求，会发现具有如下结构的响应，作为模型生成的核心内容（“choices”->“message”->“content”）证实以JSON形式返回的。
{
"choices": [
{
"content_filter_results": {…},
"finish_reason": "stop",
"index": 0,
"logprobs": null,
"message": {
"annotations": [],
"content": "{\\"name\\":\\"Jayden\\",\\"age\\":18,\\"hobbies\\":[\\"hiking\\",\\"painting\\"]}",
"refusal": null,
"role": "assistant"
}
}
],
"created": 1772427776,
"id": "chatcmpl-DEpiicspzw02JInBNnNkd1GGd05OK",
"model": "gpt-5.2-chat-2025-12-11",
"object": "chat.completion",
"prompt_filter_results": [
{…}
],
"system_fingerprint": null,
"usage": {…},
}
JsonOutputParser按照如下的方式重写了parse_result方法，它将带解析文本从第一个Generation对象提取出来后，会调用parse_json_markdown函数对其解析并生成具有目标结构的字典。从方法命名不难看出，此方法将待解析内容视为Markdown文本中的一个JSON片段。如果partial参数的值为True，以为者得到的文本可能不完整，所以即使解析失败也不会抛出异常，而是返回None。重写的parse方法会转而调用重写的parse_result。
class JsonOutputParser(BaseCumulativeTransformOutputParser[Any]):
@override
def _diff(self, prev: Any | None, next: Any) –> Any:
return jsonpatch.make_patch(prev, next).patch
@override
def parse_result(self, result: list[Generation], *, partial: bool = False) –> Any:
text = result[0].text
text = text.strip()
if partial:
try:
return parse_json_markdown(text)
except JSONDecodeError:
return None
else:
try:
return parse_json_markdown(text)
except JSONDecodeError as e:
msg = f"Invalid json output: {text}"
raise OutputParserException(msg, llm_output=text) from e
def parse(self, text: str) –> Any:
return self.parse_result([Generation(text=text)])
def get_format_instructions(self) –> str
通过前面实例的演示，我们知道get_format_instructions方法输出的指令并不是仅仅告诉模型按照JSON格式输出生成的内容，它还对JSON格式输出提出了一些要求或者约束。除此之外，它还根据我们指定的解析类型输出了对应的Schema和实例示范。JsonOutputParser还通过重写_diff方法提供了针对差分模式的支持，具体是调用jsonpatch包提供的make_patch方法。该遵循RFC 6902（JSON Patch）标准，作用是比较两个JSON对象，生成一系列指令（操作列表），描述如何通过最小步骤把第一个对象变成第二个对象。make_patch函数是为了高效传输结构化数据的变更而生的。在长文本或复杂JSON生成场景下，它是提升系统响应性能的神器。
