Python LangChain-Core源码解析(Message、Generation、BaseLanguageModel、BaseLLMOutputParser)

# 消息标识 id: str | None = Field(default=None, coerce_numbers_to_str=True)# 消息类型 type: str# 消息名称 name: str | None = None# 消息内容 content: str | list[str | dict]"""The contents of the message."""# 附加参数 additional_kwargs: dict = Field(default_factory=dict)"""Reserved for additional payload data associated with the message."""# 响应元数据 response_metadata: dict = Field(default_factory=dict)"""Examples: response headers, logprobs, token counts, model name."""# Pydantic模型行为配置model_config = ConfigDict(      extra="allow",)"""A TypedDict for configuring Pydantic behaviour."""

# 拼接运算符重载 def __add__(self, other: Any) -> ChatPromptTemplate:    """Concatenate this message with another message."""# Message合并工具 def merge_content(  first_content: str | list[str | dict],      *contents: str | list[str | dict],) -> str | list[str | dict]:    """Merge multiple message contents.

-> user_input      # 用户输入-> HumanMessage    # 封装Message-> ChatModel       # 模型-> ChatGeneration  # 模型输出-> AIMessage       # 获取模型Message

• 纯文本补全模型（如GPT-3）
  输出LLMResult，包含Generation。

• 对话模型（如GPT-3.5/4）
  输出ChatResult，包含ChatGeneration。
• 流式输出
  会先返回GenerationChunk或
  ChatGenerationChunk，最终拼接成完整的Generation或ChatGneration。

• LanguageModelInput：模型输入类型，支持str、ChatMessage列表、PromptValue等；
• RunnableConfig：配置项，支持设置temperature、max_tokens、timeout等模型参数，以及tags、metadata等追踪信息。

场景 1：解析为 JSON（基础结构化）

使用 JsonOutputParser（BaseLLMOutputParser的子类）将 LLM 输出解析为 Python 字典：

from langchain_core.output_parsers import JsonOutputParserfrom langchain_core.prompts import PromptTemplatefrom langchain_openai import ChatOpenAI# 1. 初始化解析器json_parser = JsonOutputParser()# 2. 构建提示词（加入格式说明，确保LLM输出JSON）prompt = PromptTemplate(    template="回答以下问题，并严格输出JSON格式：{question}\n{format_instructions}",    input_variables=["question"],    partial_variables={"format_instructions": json_parser.get_format_instructions()})# 3. 初始化LLM并调用llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)chain = prompt | llm | json_parser# 4. 执行并获取结构化结果result = chain.invoke({"question": "介绍苹果公司：名称、成立时间、创始人"})print(type(result))  # <class 'dict'>print(result)# 示例输出：{"名称": "苹果公司", "成立时间": "1976年4月1日", "创始人": "史蒂夫·乔布斯、史蒂夫·沃兹尼亚克、罗纳德·韦恩"}

场景 2：解析为 Pydantic 模型（强类型结构化）

这是最佳实践核心，通过 Pydantic 模型定义结构化 schema，确保解析结果的类型安全：

from pydantic import BaseModel, Fieldfrom langchain_core.output_parsers import PydanticOutputParser# 1. 定义Pydantic模型（强类型schema）class CompanyInfo(BaseModel):    name: str = Field(description="公司名称")    founded: str = Field(description="成立时间")    founders: list[str] = Field(description="创始人列表")# 2. 初始化解析器（自动生成格式说明）parser = PydanticOutputParser(pydantic_object=CompanyInfo)# 3. 构建提示词prompt = PromptTemplate(    template="回答以下问题，严格按照指定格式输出：{question}\n格式要求：{format_instructions}",    input_variables=["question"],    partial_variables={"format_instructions": parser.get_format_instructions()})# 4. 执行链llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)chain = prompt | llm | parser# 5. 解析结果（直接得到Pydantic对象）result = chain.invoke({"question": "介绍苹果公司：名称、成立时间、创始人"})print(type(result))  # <class '__main__.CompanyInfo'>print(result.name)   # 苹果公司print(result.founders)  # ["史蒂夫·乔布斯", "史蒂夫·沃兹尼亚克", "罗纳德·韦恩"]

场景 3：使用StructuredOutputParser

     对于多字段、复杂结构，可使用 StructuredOutputParser简化 schema 定义：

from langchain_core.output_parsers import StructuredOutputParser, ResponseSchema# 定义响应schemaresponse_schemas = [    ResponseSchema(name="name", description="公司名称"),    ResponseSchema(name="founded", description="成立时间"),    ResponseSchema(name="founders", description="创始人列表，用数组表示")]parser = StructuredOutputParser.from_response_schemas(response_schemas)# 后续用法与Pydantic解析器一致

场景 4：自定义解析器（适配特殊输出格式）

若 LLM 输出非标准格式（如 “键值对：xxx”），可继承 BaseLLMOutputParser实现自定义解析：

from langchain_core.output_parsers import BaseLLMOutputParserfrom langchain_core.exceptions import OutputParserException# 自定义解析器：解析 "名称：苹果；成立时间：1976；创始人：乔布斯" 格式class CustomKVParser(BaseLLMOutputParser):    def parse(self, text: str) -> dict:        try:            result = {}            # 按分号分割键值对            pairs = text.strip().split("；")            for pair in pairs:                if "：" in pair:                    key, value = pair.split("：", 1)                    result[key.strip()] = value.strip()            return result        except Exception as e:            raise OutputParserException(f"解析失败：{e}，原始文本：{text}")# 使用自定义解析器parser = CustomKVParser()llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)chain = PromptTemplate.from_template("用【键：值；】格式介绍苹果公司：名称、成立时间、创始人") | llm | parserresult = chain.invoke({})print(result)  # {"名称": "苹果公司", "成立时间": "1976年4月1日", "创始人": "乔布斯"}