flowchart LR
    PROMPT[PromptTemplate] --> MODEL[LanguageModel]
    MODEL --> PARSER[OutputParser]
    PARSER --> APP[Application Logic]

    style PARSER fill:#fff4e1

flowchart TB
    subgraph External["外部调用层"]
        USER[用户应用]
        CHAIN[LCEL Chain]
    end

    subgraph Interface["Runnable接口层"]
        INVOKE[invoke/ainvoke]
        TRANSFORM[transform/atransform]
        STREAM[stream/astream]
    end

    subgraph Core["核心解析层"]
        PARSE_RESULT[parse_result]
        PARSE[parse]
        FORMAT_INST[get_format_instructions]
    end

    subgraph Implementation["解析器实现层"]
        direction LR
        subgraph Simple["简单解析器"]
            STR[StrOutputParser]
        end

        subgraph Structured["结构化解析器"]
            JSON[JsonOutputParser]
            PYDANTIC[PydanticOutputParser]
        end

        subgraph List["列表解析器"]
            CSV[CommaSeparatedList]
            NUMBERED[NumberedList]
            MARKDOWN[MarkdownList]
        end
    end

    subgraph Base["基础抽象层"]
        BASE_OUTPUT[BaseOutputParser]
        BASE_TRANSFORM[BaseTransformOutputParser]
        BASE_CUMULATIVE[BaseCumulativeTransformOutputParser]
        BASE_GEN[BaseGenerationOutputParser]
        BASE_LLM[BaseLLMOutputParser]
    end

    subgraph Dependencies["依赖模块"]
        RUNNABLE[Runnables模块]
        MESSAGES[Messages模块]
        OUTPUTS[Outputs模块]
        UTILS[Utils工具]
    end

    USER --> CHAIN
    CHAIN --> INVOKE
    CHAIN --> TRANSFORM
    CHAIN --> STREAM

    INVOKE --> PARSE_RESULT
    TRANSFORM --> PARSE_RESULT
    STREAM --> PARSE_RESULT

    PARSE_RESULT --> PARSE
    PARSE --> Implementation

    STR --> BASE_TRANSFORM
    JSON --> BASE_CUMULATIVE
    PYDANTIC --> JSON
    CSV --> BASE_TRANSFORM
    NUMBERED --> BASE_TRANSFORM
    MARKDOWN --> BASE_TRANSFORM

    BASE_TRANSFORM --> BASE_OUTPUT
    BASE_CUMULATIVE --> BASE_TRANSFORM
    BASE_OUTPUT --> BASE_GEN
    BASE_GEN --> BASE_LLM
    BASE_OUTPUT -.继承.-> RUNNABLE

    Core -.调用.-> MESSAGES
    Core -.调用.-> OUTPUTS
    Core -.调用.-> UTILS

    style Core fill:#fff4e1
    style Base fill:#e1f5ff
    style Implementation fill:#e8f5e9

flowchart TB
    subgraph Base["基类层"]
        direction TB
        BASE_LLM[BaseLLMOutputParser<br/>最基础抽象]
        BASE_GEN[BaseGenerationOutputParser<br/>Generation解析]
        BASE_PARSER[BaseOutputParser<br/>核心基类]
        TRANSFORM_PARSER[BaseTransformOutputParser<br/>流式解析]
        CUMULATIVE_PARSER[BaseCumulativeTransformOutputParser<br/>累积解析]

        BASE_LLM --> BASE_GEN
        BASE_LLM --> BASE_PARSER
        BASE_PARSER --> TRANSFORM_PARSER
        TRANSFORM_PARSER --> CUMULATIVE_PARSER
    end

    subgraph Common["常用解析器"]
        STR_PARSER[StrOutputParser<br/>字符串解析]
        JSON_PARSER[JsonOutputParser<br/>JSON解析]
        PYDANTIC_PARSER[PydanticOutputParser<br/>Pydantic模型]
    end

    subgraph Specialized["列表解析器"]
        LIST_BASE[ListOutputParser<br/>列表基类]
        CSV_PARSER[CommaSeparatedListOutputParser<br/>CSV列表]
        NUM_PARSER[NumberedListOutputParser<br/>编号列表]
        MD_PARSER[MarkdownListOutputParser<br/>Markdown列表]

        LIST_BASE --> CSV_PARSER
        LIST_BASE --> NUM_PARSER
        LIST_BASE --> MD_PARSER
    end

    subgraph Tools["工具相关"]
        JSON_TOOLS[JsonOutputKeyToolsParser<br/>JSON工具]
        PYDANTIC_TOOLS[PydanticToolsParser<br/>Pydantic工具]
    end

    TRANSFORM_PARSER -.-> STR_PARSER
    CUMULATIVE_PARSER -.-> JSON_PARSER
    JSON_PARSER -.-> PYDANTIC_PARSER
    TRANSFORM_PARSER -.-> LIST_BASE
    BASE_PARSER -.-> JSON_TOOLS
    BASE_PARSER -.-> PYDANTIC_TOOLS

    style BASE_PARSER fill:#e1f5ff
    style JSON_PARSER fill:#fff4e1
    style PYDANTIC_PARSER fill:#e8f5e9
    style CUMULATIVE_PARSER fill:#ffe1e1

classDiagram
    class BaseLLMOutputParser~T~ {
        <<abstract>>
        +parse_result(result: List[Generation], partial: bool)* T
        +aparse_result(result: List[Generation], partial: bool) T
    }

    class BaseOutputParser~T~ {
        <<abstract>>
        +parse(text: str)* T
        +aparse(text: str) T
        +parse_result(result: List[Generation], partial: bool) T
        +parse_with_prompt(text: str, prompt: PromptValue) T
        +get_format_instructions() str
        +invoke(input: str|BaseMessage, config: RunnableConfig) T
        +ainvoke(input: str|BaseMessage, config: RunnableConfig) T
        +_type: str
        +InputType: Type
        +OutputType: Type[T]
    }

    class BaseTransformOutputParser~T~ {
        +transform(input: Iterator, config: RunnableConfig) Iterator[T]
        +atransform(input: AsyncIterator, config: RunnableConfig) AsyncIterator[T]
        #_transform(input: Iterator) Iterator[T]
        #_atransform(input: AsyncIterator) AsyncIterator[T]
    }

    class BaseCumulativeTransformOutputParser~T~ {
        +diff: bool
        #_diff(prev: T|None, next: T) T
        #_transform(input: Iterator) Iterator[T]
        #_atransform(input: AsyncIterator) AsyncIterator[T]
    }

    class StrOutputParser {
        +parse(text: str) str
        +_type: str = "default"
    }

    class JsonOutputParser {
        +pydantic_object: Type[BaseModel]|None
        +parse(text: str) dict|list
        +parse_result(result: List[Generation], partial: bool) dict|list
        +get_format_instructions() str
        +_diff(prev: Any, next: Any) Any
        -_get_schema(pydantic_object: Type[BaseModel]) dict
    }

    class PydanticOutputParser~T~ {
        +pydantic_object: Type[T]
        +parse(text: str) T
        +parse_result(result: List[Generation], partial: bool) T|None
        +get_format_instructions() str
        +OutputType: Type[T]
        -_parse_obj(obj: dict) T
        -_parser_exception(e: Exception, json_object: dict) OutputParserException
    }

    class ListOutputParser {
        <<abstract>>
        +parse(text: str)* List[str]
        +parse_iter(text: str) Iterator[re.Match]
        +_transform(input: Iterator) Iterator[List[str]]
        +_atransform(input: AsyncIterator) AsyncIterator[List[str]]
    }

    class CommaSeparatedListOutputParser {
        +parse(text: str) List[str]
        +get_format_instructions() str
        +_type: str = "comma-separated-list"
    }

    class OutputParserException {
        <<Exception>>
        +error: str
        +observation: str|None
        +llm_output: str|None
    }

    BaseLLMOutputParser <|-- BaseOutputParser
    BaseOutputParser <|-- BaseTransformOutputParser
    BaseTransformOutputParser <|-- BaseCumulativeTransformOutputParser
    BaseTransformOutputParser <|-- StrOutputParser
    BaseCumulativeTransformOutputParser <|-- JsonOutputParser
    JsonOutputParser <|-- PydanticOutputParser
    BaseTransformOutputParser <|-- ListOutputParser
    ListOutputParser <|-- CommaSeparatedListOutputParser
    BaseOutputParser ..> OutputParserException : raises

    note for BaseOutputParser "继承RunnableSerializable<br/>实现Runnable协议"
    note for BaseCumulativeTransformOutputParser "累积输入块<br/>支持diff模式"

sequenceDiagram
    autonumber
    participant User as 用户代码
    participant Chain as LCEL Chain
    participant Model as LanguageModel
    participant Parser as OutputParser
    participant Core as 核心解析逻辑

    Note over User,Core: 场景1: 标准同步调用
    User->>Chain: chain.invoke(input)
    Chain->>Model: model.invoke(messages)
    Model-->>Chain: AIMessage
    Chain->>Parser: parser.invoke(message)
    activate Parser
    Parser->>Parser: 类型判断(str vs BaseMessage)
    Parser->>Core: _call_with_config(parse_result)
    activate Core
    Core->>Core: 包装为Generation
    Core->>Core: parse_result([generation])
    Core->>Core: parse(text)
    Core->>Core: 具体解析逻辑
    Core-->>Parser: 结构化结果
    deactivate Core
    Parser-->>Chain: 结构化结果
    deactivate Parser
    Chain-->>User: 最终结果

    Note over User,Core: 场景2: 流式调用
    User->>Chain: chain.stream(input)
    Chain->>Model: model.stream(messages)
    loop 每个chunk
        Model-->>Chain: BaseMessageChunk
        Chain->>Parser: parser.transform(iter)
        activate Parser
        Parser->>Core: _transform_stream_with_config
        activate Core
        Core->>Core: 累积chunk(如果是Cumulative)
        Core->>Core: parse_result([acc_gen], partial=True)
        Core-->>Parser: 部分结果(如果有变化)
        deactivate Core
        Parser-->>Chain: yield 部分结果
        deactivate Parser
        Chain-->>User: yield 部分结果
    end

    Note over User,Core: 场景3: 异步调用
    User->>Chain: await chain.ainvoke(input)
    Chain->>Model: await model.ainvoke(messages)
    Model-->>Chain: AIMessage
    Chain->>Parser: await parser.ainvoke(message)
    activate Parser
    Parser->>Core: _acall_with_config(aparse_result)
    activate Core
    Core->>Core: await aparse_result([generation])
    Core->>Core: await run_in_executor(parse, text)
    Core->>Core: 在线程池中执行parse
    Core-->>Parser: 结构化结果
    deactivate Core
    Parser-->>Chain: 结构化结果
    deactivate Parser
    Chain-->>User: 最终结果

用户代码
  ↓
LCEL Chain (pipe操作符组装)
  ↓
Runnable接口 (invoke/transform/stream)
  ↓
BaseOutputParser.invoke/transform
  ↓
_call_with_config (配置管理+回调)
  ↓
parse_result (Generation → 文本)
  ↓
parse (具体解析逻辑)
  ↓
结构化结果

用户调用
  ↓
parser.invoke(input: str | BaseMessage)  [BaseOutputParser继承]
  ↓
_call_with_config(lambda: parse_result([generation]))
  ↓
parse_result([generation])  [BaseOutputParser实现]
  ↓
parse(generation[0].text)  [StrOutputParser实现]
  ↓
return text (直接返回)

sequenceDiagram
    autonumber
    participant User as 用户代码
    participant Parser as StrOutputParser
    participant Base as BaseOutputParser
    participant Config as RunnableConfig

    Note over User,Config: 场景1: 同步调用
    User->>Parser: invoke("Hello")
    activate Parser
    Parser->>Base: invoke(input, config)
    activate Base
    Base->>Base: 判断类型: isinstance(input, BaseMessage)?
    alt input是str
        Base->>Config: _call_with_config(parse_result)
        activate Config
        Config->>Base: parse_result([Generation(text="Hello")])
        Base->>Parser: parse("Hello")
        activate Parser
        Parser->>Parser: return text
        Parser-->>Base: "Hello"
        deactivate Parser
        Base-->>Config: "Hello"
        deactivate Config
        Config-->>Parser: "Hello"
    else input是BaseMessage
        Base->>Config: _call_with_config(parse_result)
        Config->>Base: parse_result([ChatGeneration(message=input)])
        Base->>Parser: parse(message.content)
        Parser-->>Base: content string
        Base-->>Config: content string
        Config-->>Parser: content string
    end
    deactivate Base
    Parser-->>User: "Hello"
    deactivate Parser

    Note over User,Config: 场景2: 流式调用
    User->>Parser: transform(iter(["Hello", " ", "World"]))
    activate Parser
    Parser->>Base: transform(input, config)
    activate Base
    Base->>Config: _transform_stream_with_config(_transform, ...)
    activate Config
    loop 每个chunk
        Config->>Parser: _transform(["Hello"])
        activate Parser
        Parser->>Base: parse_result([Generation(text="Hello")])
        Base->>Parser: parse("Hello")
        Parser->>Parser: return text
        Parser-->>Config: yield "Hello"
        deactivate Parser
    end
    Config-->>Base: Iterator[str]
    deactivate Config
    Base-->>Parser: Iterator[str]
    deactivate Base
    Parser-->>User: Iterator["Hello", " ", "World"]
    deactivate Parser

from langchain_core.output_parsers.transform import BaseTransformOutputParser

class StrOutputParser(BaseTransformOutputParser[str]):
    """字符串输出解析器

    最简单的解析器,直接返回文本内容。
    继承BaseTransformOutputParser以支持流式处理。
    """

    def parse(self, text: str) -> str:
        """直接返回输入文本,不做任何处理

        Args:
            text: 输入文本

        Returns:
            原始文本
        """
        return text

    @property
    def _type(self) -> str:
        """序列化类型标识"""
        return "default"

class BaseTransformOutputParser(BaseOutputParser[T]):
    def _transform(self, input: Iterator[str | BaseMessage]) -> Iterator[T]:
        """逐块处理输入

        对每个chunk调用parse_result,立即yield结果。
        适用于不需要累积的解析器(如StrOutputParser)。
        """
        for chunk in input:
            if isinstance(chunk, BaseMessage):
                # 从BaseMessage提取content
                yield self.parse_result([ChatGeneration(message=chunk)])
            else:
                # 字符串直接包装为Generation
                yield self.parse_result([Generation(text=chunk)])

class BaseOutputParser(BaseLLMOutputParser, RunnableSerializable[LanguageModelOutput, T]):
    def invoke(self, input: str | BaseMessage, config: RunnableConfig | None = None) -> T:
        """Runnable接口的invoke实现

        Args:
            input: 字符串或BaseMessage
            config: 可选的运行时配置(回调、tags等)

        Returns:
            解析后的结构化结果
        """
        if isinstance(input, BaseMessage):
            # 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",
        )

    def parse_result(self, result: list[Generation], *, partial: bool = False) -> T:
        """从Generation列表解析

        默认实现: 提取第一个Generation的text字段,调用parse。

        Args:
            result: Generation列表(通常只有一个)
            partial: 是否为部分结果(流式模式)

        Returns:
            解析结果
        """
        return self.parse(result[0].text)

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_template("讲一个关于{topic}的笑话")
model = ChatOpenAI()
parser = StrOutputParser()

chain = prompt | model | parser

# 同步调用
result = chain.invoke({"topic": "程序员"})
print(result)  # "为什么程序员..."(字符串)

# 流式调用
for chunk in chain.stream({"topic": "程序员"}):
    print(chunk, end="", flush=True)  # 逐字输出

用户调用
  ↓
parser.invoke(input: str | BaseMessage)  [BaseOutputParser继承]
  ↓
_call_with_config(lambda: parse_result([generation]))
  ↓
parse_result([generation], partial=False)  [JsonOutputParser覆盖]
  ↓
text = generation[0].text.strip()
  ↓
parse_json_markdown(text)  [工具函数]
  ↓  (尝试提取markdown代码块中的JSON)
json.loads(extracted_json)
  ↓
(可选) pydantic_object.model_validate(parsed)
  ↓
return dict | list

用户调用 chain.stream()
  ↓
parser.transform(Iterator[str | BaseMessage])
  ↓
BaseCumulativeTransformOutputParser._transform
  ↓
loop: 累积chunk → acc_gen
  ↓
parse_result([acc_gen], partial=True)
  ↓
parse_json_markdown(text) - 尝试解析
  ↓  (成功)
if parsed != prev_parsed:
    if diff: yield JSONPatch(prev, parsed)
    else: yield parsed

sequenceDiagram
    autonumber
    participant User as 用户代码
    participant Parser as JsonOutputParser
    participant Cumulative as BaseCumulativeTransform
    participant Utils as JSON工具
    participant Pydantic as Pydantic验证

    Note over User,Pydantic: 场景1: 非流式解析
    User->>Parser: invoke('{"name": "Alice", "age": 30}')
    activate Parser
    Parser->>Parser: parse_result([generation], partial=False)
    activate Parser
    Parser->>Parser: text = result[0].text.strip()
    Parser->>Utils: parse_json_markdown(text)
    activate Utils
    Utils->>Utils: 检测markdown代码块(```json...)
    Utils->>Utils: 提取JSON部分
    Utils->>Utils: json.loads(json_str)
    Utils-->>Parser: {"name": "Alice", "age": 30}
    deactivate Utils
    alt pydantic_object存在
        Parser->>Pydantic: model_validate(parsed)
        activate Pydantic
        Pydantic->>Pydantic: 验证字段类型和约束
        Pydantic-->>Parser: 验证通过/失败
        deactivate Pydantic
    end
    Parser-->>Parser: parsed dict
    deactivate Parser
    Parser-->>User: {"name": "Alice", "age": 30}
    deactivate Parser

    Note over User,Pydantic: 场景2: 流式解析(累积模式)
    User->>Parser: transform(iter(['{"na', 'me": "A', 'lice"}']))
    activate Parser
    Parser->>Cumulative: _transform(input)
    activate Cumulative
    loop 每个chunk
        Cumulative->>Cumulative: chunk_gen = GenerationChunk(text=chunk)
        Cumulative->>Cumulative: acc_gen = acc_gen + chunk_gen
        Note right of Cumulative: 累积: '{"na' → '{"name": "A' → '{"name": "Alice"}'
        Cumulative->>Parser: parse_result([acc_gen], partial=True)
        activate Parser
        Parser->>Utils: parse_json_markdown(accumulated_text)
        activate Utils
        alt JSON完整
            Utils->>Utils: json.loads成功
            Utils-->>Parser: parsed dict
            Parser->>Parser: if parsed != prev_parsed
            Parser-->>Cumulative: yield parsed
        else JSON不完整
            Utils->>Utils: JSONDecodeError
            Utils-->>Parser: return None
            Parser-->>Cumulative: 不yield(等待更多数据)
        end
        deactivate Utils
        deactivate Parser
    end
    Cumulative-->>Parser: Iterator[dict]
    deactivate Cumulative
    Parser-->>User: yield完整JSON
    deactivate Parser

    Note over User,Pydantic: 场景3: diff模式(增量输出)
    User->>Parser: parser.diff = True<br/>transform(iter(['{"items"', ': ["a"],', ' "count": 1}']))
    activate Parser
    Parser->>Cumulative: _transform(input)
    activate Cumulative
    loop 每个chunk
        Cumulative->>Cumulative: 累积chunk
        Cumulative->>Parser: parse_result([acc_gen], partial=True)
        Parser-->>Cumulative: parsed (new version)
        Cumulative->>Cumulative: _diff(prev_parsed, parsed)
        Note right of Cumulative: JSONPatch: [<br/>  {"op": "add", "path": "/count", "value": 1}<br/>]
        Cumulative-->>Parser: yield JSONPatch operations
    end
    deactivate Cumulative
    Parser-->>User: yield增量变化
    deactivate Parser

    Note over User,Pydantic: 场景4: 解析失败
    User->>Parser: invoke('Invalid JSON{')
    activate Parser
    Parser->>Parser: parse_result([generation], partial=False)
    Parser->>Utils: parse_json_markdown('Invalid JSON{')
    activate Utils
    Utils->>Utils: json.loads(...)
    Utils->>Utils: JSONDecodeError
    Utils-->>Parser: raise Exception
    deactivate Utils
    Parser->>Parser: catch JSONDecodeError
    Parser->>Parser: raise OutputParserException
    Parser-->>User: OutputParserException:<br/>llm_output='Invalid JSON{'
    deactivate Parser

from langchain_core.output_parsers.transform import BaseCumulativeTransformOutputParser
from langchain_core.utils.json import parse_json_markdown
import json
import jsonpatch

class JsonOutputParser(BaseCumulativeTransformOutputParser[Any]):
    """JSON输出解析器

    支持流式模式和JSONPatch增量输出。
    """

    pydantic_object: type[BaseModel] | None = None
    """可选的Pydantic模型(用于生成schema和验证)"""

    def parse_result(self, result: list[Generation], *, partial: bool = False) -> Any:
        """从Generation列表解析JSON

        Args:
            result: Generation列表
            partial: 是否为部分结果(流式模式)

        Returns:
            解析后的dict/list,或None(partial模式下解析失败)

        Raises:
            OutputParserException: partial=False且解析失败时
        """
        text = result[0].text.strip()

        if partial:
            # 流式模式: 解析失败返回None
            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:
        """便捷方法: 从文本解析JSON"""
        return self.parse_result([Generation(text=text)])

    def get_format_instructions(self) -> str:
        """生成格式指令"""
        if self.pydantic_object is None:
            return "Return a JSON object."

        # 从Pydantic模型生成schema
        schema = self.pydantic_object.model_json_schema()
        # (省略schema简化逻辑)
        schema_str = json.dumps(schema, ensure_ascii=False)
        return JSON_FORMAT_INSTRUCTIONS.format(schema=schema_str)

    def _diff(self, prev: Any | None, next: Any) -> Any:
        """计算JSONPatch增量

        Args:
            prev: 上一次的解析结果
            next: 当前解析结果

        Returns:
            JSONPatch操作列表
        """
        return jsonpatch.make_patch(prev, next).patch

def parse_json_markdown(text: str) -> Any:
    """从文本中提取并解析JSON

    支持:
    - 纯JSON: {"key": "value"}
    - Markdown代码块: ```json\n{"key": "value"}\n```
    - 普通代码块: ```\n{"key": "value"}\n```

    Args:
        text: 输入文本

    Returns:
        解析后的JSON对象

    Raises:
        JSONDecodeError: 解析失败
    """
    # 1) 尝试提取markdown代码块
    if "```json" in text:
        text = text.split("```json")[1].split("```")[0]
    elif "```" in text:
        text = text.split("```")[1].split("```")[0]

    # 2) 清理空白
    text = text.strip()

    # 3) 解析JSON
    return json.loads(text)

class BaseCumulativeTransformOutputParser(BaseTransformOutputParser[T]):
    diff: bool = False
    """是否输出增量差异"""

    def _transform(self, input: Iterator[str | BaseMessage]) -> Iterator[Any]:
        """累积输入并增量解析

        关键逻辑:
        1. 累积每个chunk到acc_gen
        2. 尝试解析累积结果(partial=True)
        3. 如果解析成功且结果有变化,yield结果
        4. 如果diff=True,yield增量而非完整结果
        """
        prev_parsed = None
        acc_gen: GenerationChunk | ChatGenerationChunk | None = None

        for chunk in input:
            # 1) 包装chunk
            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)

            # 2) 累积
            acc_gen = chunk_gen if acc_gen is None else acc_gen + chunk_gen

            # 3) 尝试解析
            parsed = self.parse_result([acc_gen], partial=True)

            # 4) 检查变化
            if parsed is not None and parsed != prev_parsed:
                if self.diff:
                    # 输出增量
                    yield self._diff(prev_parsed, parsed)
                else:
                    # 输出完整结果
                    yield parsed
                prev_parsed = parsed

from pydantic import BaseModel, Field

class Joke(BaseModel):
    setup: str = Field(description="笑话的铺垫")
    punchline: str = Field(description="笑点")
    rating: int = Field(description="有趣程度1-10")

# 创建解析器(带schema)
parser = JsonOutputParser(pydantic_object=Joke)

# 构建链
prompt = ChatPromptTemplate.from_template(
    "讲一个笑话,{format_instructions}\n\n主题: {topic}"
)
chain = prompt | model | parser

# 使用
result = chain.invoke({
    "topic": "程序员",
    "format_instructions": parser.get_format_instructions()
})

print(result)
# {'setup': '...', 'punchline': '...', 'rating': 8}

# 流式调用
for partial_result in chain.stream({"topic": "程序员", ...}):
    print(f"部分结果: {partial_result}")

# diff模式
parser.diff = True
for patch in chain.stream({"topic": "程序员", ...}):
    print(f"增量变化: {patch}")
    # [{"op": "add", "path": "/setup", "value": "..."}]

from pydantic import BaseModel, Field

class Article(BaseModel):
    title: str = Field(description="文章标题")
    summary: str = Field(description="文章摘要")
    keywords: list[str] = Field(description="关键词列表")
    category: str = Field(description="文章分类")

parser = PydanticOutputParser(pydantic_object=Article)

prompt = ChatPromptTemplate.from_template(
    """分析以下文章并提取信息。

{format_instructions}

文章内容:
{content}"""
)

chain = prompt | model | parser

article_info = chain.invoke({
    "content": "长篇文章内容...",
    "format_instructions": parser.get_format_instructions()
})

# 第一步: 生成JSON
json_parser = JsonOutputParser()
step1 = prompt1 | model | json_parser

# 第二步: 验证Pydantic
pydantic_parser = PydanticOutputParser(pydantic_object=MyModel)
step2 = RunnableLambda(lambda x: pydantic_parser.parse(json.dumps(x)))

# 组合
chain = step1 | step2

from langchain_core.output_parsers import OutputFixingParser

# 基础解析器
base_parser = PydanticOutputParser(pydantic_object=MyModel)

# 包装为修复解析器
fixing_parser = OutputFixingParser.from_llm(
    parser=base_parser,
    llm=model
)

# 即使输出格式有误,也会尝试修复
result = fixing_parser.parse(malformed_output)

# 简单字符串: StrOutputParser
chain = prompt | model | StrOutputParser()

# 简单JSON: JsonOutputParser
chain = prompt | model | JsonOutputParser()

# 需要验证: PydanticOutputParser
chain = prompt | model | PydanticOutputParser(pydantic_object=MyModel)

# 推荐: 在prompt中包含格式指令
parser = PydanticOutputParser(pydantic_object=MyModel)

prompt = ChatPromptTemplate.from_template(
    """任务: {task}

{format_instructions}

输入: {input}"""
)

chain = (
    {"task": ..., "input": ..., "format_instructions": lambda _: parser.get_format_instructions()}
    | prompt
    | model
    | parser
)

# 推荐: 捕获OutputParserException
from langchain_core.exceptions import OutputParserException

try:
    result = parser.parse(output)
except OutputParserException as e:
    print(f"解析失败: {e}")
    print(f"原始输出: {e.llm_output}")
    # 降级处理或重试
    result = fallback_value