LangChain Core 源码学习路线：第 3 篇 OutputParser 与结构化输出源码解剖#

1
from pydantic import BaseModel, Field
2
from langchain.chat_models import init_chat_model
3

4
class TravelIntent(BaseModel):
5
    destination: str | None = Field(description="目的地")
6
    days: int | None = Field(description="旅行天数")
7
    preferences: list[str] = Field(description="用户偏好")
8
    need_clarification: bool = Field(description="是否需要追问")
9

10
model = init_chat_model("openai:gpt-4.1-mini")
11
structured_model = model.with_structured_output(TravelIntent)
12

13
result = structured_model.invoke(
14
    "我想去东京玩 5 天，喜欢美食和城市漫步"
15
)
16

17
print(result)

1
TravelIntent(
2
    destination="东京",
3
    days=5,
4
    preferences=["美食", "城市漫步"],
5
    need_clarification=False,
6
)

1
result 不再是 AIMessage；
2
也不是字符串；
3
而是 Pydantic 对象。

1
raw_result = model.invoke(
2
    "我想去东京玩 5 天，喜欢美食和城市漫步"
3
)
4

5
print(type(raw_result))
6
print(raw_result.content)

1
AIMessage

1
TravelIntent

1
with_structured_output
2
并不是改变模型本体；
3
而是返回了一个新的 Runnable wrapper。

1
from langchain_core.prompts import ChatPromptTemplate
2
from langchain_core.output_parsers import StrOutputParser
3

4
prompt = ChatPromptTemplate.from_messages([
5
    ("system", "你是旅行规划助手。"),
6
    ("user", "请为 {destination} 规划 {days} 天行程。")
7
])
8

9
chain = prompt | model | StrOutputParser()
10

11
result = chain.invoke({
12
    "destination": "东京",
13
    "days": 5,
14
})
15

16
print(type(result))
17
print(result)

1
str

1
AIMessage
2
  ↓
3
提取文本 content
4
  ↓
5
str

1
from langchain_core.output_parsers import JsonOutputParser
2

3
parser = JsonOutputParser()
4

5
chain = prompt | model | parser
6

7
result = chain.invoke({
8
    "destination": "东京",
9
    "days": 5,
10
})
11

12
print(type(result))
13
print(result)

1
dict 或 list

1
from langchain_core.output_parsers import PydanticOutputParser
2

3
parser = PydanticOutputParser(
4
    pydantic_object=TravelIntent
5
)
6

7
format_instructions = parser.get_format_instructions()
8

9
prompt = ChatPromptTemplate.from_messages([
10
    ("system", "你是旅行需求解析器。请严格按格式输出。"),
11
    ("user", "{input}\n\n{format_instructions}")
12
])
13

14
chain = prompt | model | parser
15

16
result = chain.invoke({
17
    "input": "我想去东京玩 5 天，喜欢美食和城市漫步",
18
    "format_instructions": format_instructions,
19
})
20

21
print(type(result))
22
print(result)

1
Prompt 中塞格式要求
2
  ↓
3
模型输出 JSON 文本
4
  ↓
5
JsonOutputParser 解析 JSON
6
  ↓
7
PydanticOutputParser 校验字段
8
  ↓
9
TravelIntent 对象

1
Pydantic schema
2
  ↓
3
转换为模型可理解的结构化输出约束
4
  ↓
5
绑定到 ChatModel wrapper
6
  ↓
7
模型生成结构化结果
8
  ↓
9
框架解析并校验
10
  ↓
11
返回 Pydantic 对象或 dict

1
TravelIntent(BaseModel)
2
  ↓
3
model.with_structured_output(TravelIntent)
4
  ↓
5
判断 schema 类型
6
  ↓
7
is_pydantic_schema = True
8
  ↓
9
选择结构化输出策略
10
  ├─ provider-native structured output
11
  └─ tool calling structured output
12
  ↓
13
构造 llm Runnable
14
  ↓
15
构造 output_parser
16
  ↓
17
返回 llm | output_parser
18
  或 RunnableMap(raw=llm) | parser_with_fallback

1
AIMessage / Generation
2
  ↓
3
取出 text
4
  ↓
5
JsonOutputParser.parse_result
6
  ↓
7
parse_json_markdown
8
  ↓
9
得到 dict
10
  ↓
11
PydanticOutputParser._parse_obj
12
  ↓
13
TravelIntent.model_validate(dict)
14
  ↓
15
返回 TravelIntent

1
模型输出文本
2
  ↓
3
去除 Markdown 代码块
4
  ↓
5
定位 JSON 对象
6
  ↓
7
json.loads / parse_partial_json
8
  ↓
9
返回 dict/list
10
  ↓
11
失败则抛 OutputParserException

1
AIMessage
2
  ↓
3
ChatGeneration
4
  ↓
5
Generation.text / message.content
6
  ↓
7
parse(text)
8
  ↓
9
返回 text 本身

1
prompt | model | parser

1
BaseOutputParser
2
  ↓
3
BaseTransformOutputParser
4
  ↓
5
RunnableSerializable

1
parser.invoke(...)
2
parser.ainvoke(...)
3
parser.batch(...)
4
parser.stream(...)
5
parser.with_retry(...)
6
parser.with_config(...)

1
with_retry
2
with_fallbacks
3
config
4
callbacks
5
tags
6
metadata

1
class BaseLLMOutputParser(Generic[T], ABC):
2

3
    @abstractmethod
4
    def parse_result(
5
        self,
6
        result: list[Generation],
7
        *,
8
        partial: bool = False,
9
    ) -> T:
10
        ...

1
list[Generation]

1
n=1
2
n=3
3
n=5

1
result[0]

1
class BaseGenerationOutputParser(
2
    BaseLLMOutputParser[T],
3
    RunnableSerializable[LanguageModelOutput, T],
4
):
5

6
    def invoke(self, input, config=None, **kwargs):
7
        if isinstance(input, BaseMessage):
8
            generation = ChatGeneration(message=input)
9
        else:
10
            generation = Generation(text=input)
11

12
        return self._call_with_config(
13
            lambda inner_input: self.parse_result(
14
                [generation]
15
            ),
16
            input,
17
            config,
18
            run_type="parser",
19
        )

1
parser.invoke(AIMessage)
2
  ↓
3
包装成 ChatGeneration
4
  ↓
5
parse_result([ChatGeneration])

1
parser.invoke("raw text")
2
  ↓
3
包装成 Generation
4
  ↓
5
parse_result([Generation])

1
AIMessage
2
str
3
Generation

1
def parse(self, text: str) -> T:
2
    ...

1
def parse_result(self, result, partial=False):
2
    return self.parse(result[0].text)

1
BaseGenerationOutputParser
2
  面向 Generation
3

4
BaseOutputParser
5
  面向 text
6

7
具体 parser
8
  实现 parse(text)

1
完整 AIMessage
2
  ↓
3
parse_result

1
AIMessageChunk
2
  ↓
3
chunk 累积
4
  ↓
5
parse_result(partial=True)
6
  ↓
7
不断产出中间解析结果

1
StrOutputParser
2
  可以天然流式输出文本
3

4
JsonOutputParser
5
  可以在 partial=True 时输出部分 JSON
6

7
PydanticOutputParser
8
  通常需要完整对象后才能稳定校验

1
请严格输出 JSON
2
不要输出解释
3
字段必须是...

1
当然可以，下面是 JSON：
2
{
3
  ...
4
}
5
希望对你有帮助！

1
def with_structured_output(
2
    self,
3
    schema: dict | type | None = None,
4
    *,
5
    method: Literal[
6
        "function_calling",
7
        "json_mode",
8
        "json_schema",
9
    ] = "function_calling",
10
    include_raw: bool = False,
11
    strict: bool | None = None,
12
    **kwargs: Any,
13
) -> Runnable[LanguageModelInput, dict | BaseModel]:
14
    ...

1
BaseChatModel 定义抽象协议；
2
具体行为通常由供应商模型类重写。

1
def with_structured_output(
2
    self,
3
    schema,
4
    *,
5
    include_raw=False,
6
    **kwargs,
7
):
8
    if kwargs:
9
        raise ValueError(
10
            "Received unsupported arguments"
11
        )
12

13
    is_pydantic_schema = is_basemodel_subclass(schema)
14

15
    llm = self.bind_tools(
16
        [schema],
17
        tool_choice="any_or_schema_name",
18
        structured_output_format={
19
            "kwargs": {...},
20
            "schema": schema,
21
        },
22
    )
23

24
    if is_pydantic_schema:
25
        output_parser = PydanticToolsParser(
26
            tools=[schema],
27
            first_tool_only=True,
28
        )
29
    else:
30
        key_name = convert_to_openai_tool(schema)[
31
            "function"
32
        ]["name"]
33

34
        output_parser = JsonOutputKeyToolsParser(
35
            key_name=key_name,
36
            first_tool_only=True,
37
        )
38

39
    if include_raw:
40
        return _include_raw_wrapper(
41
            llm,
42
            output_parser,
43
        )
44

45
    return llm | output_parser

1
{
2
    "raw": AIMessage,
3
    "parsed": TravelIntent | dict | None,
4
    "parsing_error": BaseException | None,
5
}

1
def _include_raw_wrapper(llm, output_parser):
2
    parser_assign = RunnablePassthrough.assign(
3
        parsed=itemgetter("raw") | output_parser,
4
        parsing_error=lambda _: None,
5
    )
6

7
    parser_none = RunnablePassthrough.assign(
8
        parsed=lambda _: None
9
    )
10

11
    parser_with_fallback = (
12
        parser_assign.with_fallbacks(
13
            [parser_none],
14
            exception_key="parsing_error",
15
        )
16
    )
17

18
    return RunnableMap(raw=llm) | parser_with_fallback

1
def with_structured_output_provider_native(
2
    self,
3
    schema,
4
    *,
5
    include_raw=False,
6
    strict=None,
7
):
8
    json_schema = convert_to_json_schema(schema)
9

10
    llm = self.bind(
11
        response_format={
12
            "type": "json_schema",
13
            "json_schema": json_schema,
14
            "strict": strict,
15
        },
16
        structured_output_format={
17
            "kwargs": {
18
                "method": "json_schema",
19
                "strict": strict,
20
            },
21
            "schema": schema,
22
        },
23
    )
24

25
    if is_pydantic_schema(schema):
26
        parser = PydanticOutputParser(
27
            pydantic_object=schema
28
        )
29
    else:
30
        parser = JsonOutputParser()
31

32
    if include_raw:
33
        return RunnableMap(raw=llm) | parser_with_fallback
34

35
    return llm | parser

1
tool calling 模式
2
  输出在 AIMessage.tool_calls 中
3

4
provider-native 模式
5
  输出通常在 message content 或供应商标准字段中

1
class TravelIntent(BaseModel):
2
    ...

1
{
2
  "type": "function",
3
  "function": {
4
    "name": "TravelIntent",
5
    "description": "...",
6
    "parameters": {
7
      "type": "object",
8
      "properties": {
9
        "destination": {"type": "string"},
10
        "days": {"type": "integer"}
11
      }
12
    }
13
  }
14
}

1
AIMessage(
2
    content="",
3
    tool_calls=[
4
        {
5
            "name": "TravelIntent",
6
            "args": {
7
                "destination": "东京",
8
                "days": 5,
9
            },
10
            "id": "call_...",
11
            "type": "tool_call",
12
        }
13
    ],
14
)

1
TravelIntent(...)

1
结构化输出
2
  =
3
工具调用协议的一种特殊用法

1
from pydantic import BaseModel, Field
2

3
class TravelIntent(BaseModel):
4
    destination: str | None = Field(
5
        description="目的地"
6
    )
7
    days: int | None = Field(
8
        description="旅行天数",
9
        ge=1,
10
        le=30,
11
    )
12
    preferences: list[str] = Field(
13
        description="用户偏好"
14
    )
15
    need_clarification: bool = Field(
16
        description="是否需要追问"
17
    )

1
TravelIntent.model_json_schema()

1
{
2
  "title": "TravelIntent",
3
  "type": "object",
4
  "properties": {
5
    "destination": {
6
      "anyOf": [
7
        {"type": "string"},
8
        {"type": "null"}
9
      ],
10
      "description": "目的地"
11
    },
12
    "days": {
13
      "anyOf": [
14
        {
15
          "type": "integer",
16
          "minimum": 1,
17
          "maximum": 30
18
        },
19
        {"type": "null"}
20
      ],
21
      "description": "旅行天数"
22
    },
23
    "preferences": {
24
      "type": "array",
25
      "items": {"type": "string"},
26
      "description": "用户偏好"
27
    },
28
    "need_clarification": {
29
      "type": "boolean",
30
      "description": "是否需要追问"
31
    }
32
  },
33
  "required": [
34
    "destination",
35
    "days",
36
    "preferences",
37
    "need_clarification"
38
  ]
39
}

1
Pydantic BaseModel
2
TypedDict
3
Dataclass
4
JSON Schema dict
5
OpenAI tool schema
6
Python callable

1
模型供应商可理解的 schema

1
destination: str | None

1
str
2
或
3
None

1
destination: str | None = None

1
class TravelIntent(BaseModel):
2
    destination: str | None = Field(
3
        default=None,
4
        description="目的地，未知时为 None"
5
    )

1
days: int | None = Field(
2
    default=None,
3
    ge=1,
4
    le=30,
5
    description="旅行天数，必须在 1 到 30 之间"
6
)
7

8
preferences: list[str] = Field(
9
    default_factory=list,
10
    max_length=10,
11
    description="用户偏好列表"
12
)

1
from typing import Literal
2

3
class ReflectionResult(BaseModel):
4
    risk_level: Literal["low", "medium", "high"]
5
    passed: bool
6
    issues: list[str]

1
"风险不大"
2
"一般"
3
"OK"

1
low / medium / high

1
from pydantic import model_validator
2

3
class TravelIntent(BaseModel):
4
    destination: str | None = None
5
    days: int | None = None
6
    need_clarification: bool
7

8
    @model_validator(mode="after")
9
    def check_clarification(self):
10
        if self.need_clarification:
11
            return self
12

13
        if not self.destination or not self.days:
14
            raise ValueError(
15
                "不需要追问时，destination 和 days 必须存在"
16
            )
17

18
        return self

1
最终自然语言回复
2
普通摘要
3
改写任务
4
非结构化解释
5
不需要字段校验的输出

1
意图识别
2
槽位抽取
3
计划步骤
4
工具参数
5
责任判断
6
安全决策

1
class StrOutputParser(BaseTransformOutputParser[str]):
2

3
    @property
4
    def OutputType(self) -> type[str]:
5
        return str
6

7
    def parse(self, text: str) -> str:
8
        return text

1
AIMessage
2
  ↓
3
ChatGeneration
4
  ↓
5
.text
6
  ↓
7
StrOutputParser.parse(text)
8
  ↓
9
text

1
Generation(text="...")
2
ChatGeneration(message=AIMessage(...))
3
AIMessage
4
AIMessageChunk
5
str

1
return ai_message.content

1
输入 → Generation 包装 → parse_result → parse(text)

1
chain = prompt | model | StrOutputParser()

1
chain.invoke(input)
2
  ↓
3
prompt.invoke(input)
4
  ↓
5
ChatPromptValue
6
  ↓
7
model.invoke(ChatPromptValue)
8
  ↓
9
AIMessage
10
  ↓
11
StrOutputParser.invoke(AIMessage)
12
  ↓
13
ChatGeneration(message=AIMessage)
14
  ↓
15
parse_result([ChatGeneration])
16
  ↓
17
parse(generation.text)
18
  ↓
19
str

1
返回 dict
2
返回 list
3
简单结构化输出
4
不需要 Pydantic 强校验
5
需要 streaming partial JSON

1
字段类型强约束
2
业务规则校验
3
枚举 / 范围 / 嵌套模型强约束
4
高风险业务决策

1
JsonOutputParser
2
  ↓
3
BaseCumulativeTransformOutputParser
4
  ↓
5
BaseTransformOutputParser
6
  ↓
7
BaseOutputParser

1
{"destination": "东
2
{"destination": "东京", "days":
3
{"destination": "东京", "days": 5}

1
def parse_result(
2
    self,
3
    result: list[Generation],
4
    *,
5
    partial: bool = False,
6
) -> Any:
7
    text = result[0].text
8
    text = text.strip()
9

10
    if partial:
11
        try:
12
            return parse_json_markdown(text)
13
        except JSONDecodeError:
14
            return None
15

16
    try:
17
        return parse_json_markdown(text)
18
    except JSONDecodeError as error:
19
        msg = f"Invalid json output: {text}"
20
        raise OutputParserException(
21
            msg,
22
            llm_output=text,
23
        ) from error

1
def parse(self, text: str) -> Any:
2
    return self.parse_result(
3
        [Generation(text=text)]
4
    )