系列导航：[第一篇] → [第二篇] → [第三篇] → [第四篇] → 第五篇（当前） → 第六篇：记忆模块 前置要求：已掌握前四篇内容，理解 LCEL 链的基本组合方式 本篇目标：系统掌握 LangChain 的输出解析体系。从"模型只会返回字符串"到"模型输出直接变成可用的 Python 对象"，覆盖所有内置 Parser 的原理与用法、解析失败的处理策略、自定义 Parser 的实现，以及在实际项目中的选型决策。
系列导航：[第一篇] → [第二篇] → [第三篇] → [第四篇] → 第五篇（当前） → 第六篇：记忆模块
前置要求：已掌握前四篇内容，理解 LCEL 链的基本组合方式
本篇目标：系统掌握 LangChain 的输出解析体系。从"模型只会返回字符串"到"模型输出直接变成可用的 Python 对象"，覆盖所有内置 Parser 的原理与用法、解析失败的处理策略、自定义 Parser 的实现，以及在实际项目中的选型决策。
大模型的本质是文本生成器——无论你想要什么格式的数据，模型最终吐出的都是一段字符串。
在原生调用中，你可能已经写过类似这样的代码：
import json import re response = llm.invoke("请以 JSON 格式返回用户信息：姓名、年龄、城市") raw_text = response.content # 手动解析：充满了各种边界情况的处理 try: # 模型有时会在 JSON 前后加 Markdown 代码块 if "```json" in raw_text: raw_text = re.search(r'```json\n(.*?)\n```', raw_text, re.DOTALL).group(1) elif "```" in raw_text: raw_text = re.search(r'```\n(.*?)\n```', raw_text, re.DOTALL).group(1) data = json.loads(raw_text) name = data["name"] age = data["age"] city = data["city"] except json.JSONDecodeError: # 解析失败了，怎么办？重试？返回默认值？记录日志？ print("模型返回的不是合法 JSON，原始内容：", raw_text) data = None except KeyError as e: # 字段缺失，怎么办？ print(f"缺少字段：{e}") data = None
这段代码的问题是显而易见的：
dict
LangChain 的 OutputParser 体系解决的正是这些问题。
OutputParser 在 LCEL 链中处于最末端，负责把模型的原始输出（AIMessage）转换成你需要的 Python 对象：
AIMessage
用户输入 → ChatPromptTemplate → ChatOpenAI → OutputParser → Python 对象 ↓ ↓ AIMessage(content= str / dict / "{'name':...}") Pydantic 对象
一个 OutputParser 通常由两部分组成：
1. 格式指令（Format Instructions）：告诉模型应该按什么格式输出，注入到 Prompt 里 2. 解析函数（Parse）：把模型的字符串输出解析成目标 Python 对象
from langchain_core.output_parsers import PydanticOutputParser from pydantic import BaseModel class User(BaseModel): name: str age: int parser = PydanticOutputParser(pydantic_object=User) # Parser 的两个核心方法 print(parser.get_format_instructions()) # 输出类似： # The output should be formatted as a JSON instance that conforms to the JSON schema below. # {"properties": {"name": {"type": "string"}, "age": {"type": "integer"}}, ...} # 解析模型输出 user = parser.parse('{"name": "张伟", "age": 28}') print(type(user)) # <class 'User'> print(user.name) # 张伟 print(user.age) # 28
LangChain 提供了一系列开箱即用的 Parser，覆盖绝大多数使用场景。
把 AIMessage 转成纯字符串，是最常用的 Parser，也是 LCEL 链末尾的默认选择：
from langchain_core.output_parsers import StrOutputParser from langchain_core.messages import AIMessage parser = StrOutputParser() # 接受 AIMessage result = parser.invoke(AIMessage(content="你好，我是助手")) print(result) # "你好，我是助手" print(type(result)) # <class 'str'> # 也接受纯字符串（透传） result = parser.invoke("已经是字符串了") print(result) # "已经是字符串了" # 在链中使用（最常见场景） from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate chain = ( ChatPromptTemplate.from_messages([("human", "{q}")]) | ChatOpenAI(model="gpt-4o") | StrOutputParser() # AIMessage → str ) result = chain.invoke({"q": "你好"}) print(type(result)) # <class 'str'>
专门处理模型输出 JSON 的场景。相比手动 json.loads()，它能自动处理 Markdown 代码块包裹等格式问题：
json.loads()
from langchain_core.output_parsers import JsonOutputParser from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate parser = JsonOutputParser() # 基础用法 chain = ( ChatPromptTemplate.from_messages([ ("system", "你的输出必须是合法的 JSON 格式，不要包含任何其他文字。"), ("human", "{q}"), ]) | ChatOpenAI(model="gpt-4o") | parser ) result = chain.invoke({"q": "给我一个用户信息，包含 name、age、city 字段"}) print(type(result)) # <class 'dict'> print(result["name"]) # 张三 print(result["age"]) # 28
JsonOutputParser 能自动处理以下几种模型输出格式：
JsonOutputParser
# 这些格式都能被正确解析 formats = [ '{"name": "张三", "age": 28}', # 标准 JSON '```json\n{"name": "张三", "age": 28}\n```', # Markdown 代码块 '```\n{"name": "张三", "age": 28}\n```', # 无语言标记的代码块 '以下是用户信息：\n{"name": "张三", "age": 28}', # 前有说明文字 ]
带 Schema 的 JsonOutputParser：使用 Pydantic 定义期望结构，自动生成格式指令：
from langchain_core.output_parsers import JsonOutputParser from pydantic import BaseModel, Field from typing import List class Product(BaseModel): name: str = Field(description="产品名称") price: float = Field(description="价格，单位：元") tags: List[str] = Field(description="产品标签列表") in_stock: bool = Field(description="是否有货") parser = JsonOutputParser(pydantic_object=Product) # 自动生成格式指令，注入到 Prompt 里 chain = ( ChatPromptTemplate.from_messages([ ("system", "请根据用户描述生成产品信息。\n\n{format_instructions}"), ("human", "{description}"), ]).partial(format_instructions=parser.get_format_instructions()) | ChatOpenAI(model="gpt-4o") | parser ) result = chain.invoke({"description": "一款高端蓝牙耳机，降噪效果好，售价 1299 元，现货"}) print(type(result)) # <class 'dict'>（注意：不是 Product 对象，是 dict） print(result) # {'name': '高端蓝牙耳机', 'price': 1299.0, 'tags': ['蓝牙', '降噪', '高端'], 'in_stock': True}
注意：JsonOutputParser(pydantic_object=Product) 输出的是 dict，不是 Product 实例。如果需要 Pydantic 对象，用下面的 PydanticOutputParser。
JsonOutputParser(pydantic_object=Product)
Product
PydanticOutputParser
输出直接是 Pydantic 对象，享有完整的类型检查和 IDE 自动补全：
from langchain_core.output_parsers import PydanticOutputParser from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from pydantic import BaseModel, Field, field_validator from typing import List, Optional class CodeReview(BaseModel): """代码审查结果""" language: str = Field(description="编程语言") has_bugs: bool = Field(description="是否存在 Bug") bugs: List[str] = Field(description="Bug 列表，如果没有则为空列表", default_factory=list) severity: str = Field(description="整体严重程度：low / medium / high / critical") suggestions: List[str] = Field(description="改进建议列表") score: int = Field(description="代码质量评分，0-100 分") @field_validator("severity") @classmethod def validate_severity(cls, v: str) -> str: allowed = {"low", "medium", "high", "critical"} if v.lower() not in allowed: raise ValueError(f"severity 必须是 {allowed} 之一，收到：{v}") return v.lower() @field_validator("score") @classmethod def validate_score(cls, v: int) -> int: if not 0 <= v <= 100: raise ValueError(f"score 必须在 0-100 之间，收到：{v}") return v # 创建 Parser parser = PydanticOutputParser(pydantic_object=CodeReview) # 查看自动生成的格式指令 print(parser.get_format_instructions()) # The output should be formatted as a JSON instance that conforms to the JSON schema below. # {"properties": {"language": {"description": "编程语言", "title": "Language", "type": "string"}, ...}} # ... # 构建链 chain = ( ChatPromptTemplate.from_messages([ ("system", """你是一个专业的代码审查专家。 请严格按照以下格式输出审查结果： {format_instructions}"""), ("human", "请审查以下代码：\n```\n{code}\n```"), ]).partial(format_instructions=parser.get_format_instructions()) | ChatOpenAI(model="gpt-4o", temperature=0) | parser ) code = """ def divide(a, b): return a / b def get_user(user_id): query = f"SELECT * FROM users WHERE id = {user_id}" return db.execute(query) """ result = chain.invoke({"code": code}) # result 是 CodeReview 对象，享有完整类型提示 print(type(result)) # <class 'CodeReview'> print(result.language) # Python print(result.has_bugs) # True print(result.severity) # critical print(result.score) # 35 for bug in result.bugs: print(f" - {bug}") # - 未处理除数为零的情况（ZeroDivisionError） # - SQL 注入漏洞：直接拼接用户输入到 SQL 查询 for s in result.suggestions: print(f" → {s}")
如第二篇所述，with_structured_output() 是 LangChain 1.0 推荐的结构化输出方式。它和 PydanticOutputParser 的最大区别在于底层机制：
with_structured_output()
from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field from typing import List class MovieRecommendation(BaseModel): """电影推荐结果""" title: str = Field(description="电影名称") year: int = Field(description="上映年份") genre: List[str] = Field(description="类型标签") reason: str = Field(description="推荐理由，50字以内") rating: float = Field(description="豆瓣评分，0-10") llm = ChatOpenAI(model="gpt-4o") structured_llm = llm.with_structured_output(MovieRecommendation) # 不需要任何格式指令注入，模型通过 Function Call 直接输出结构化数据 result = structured_llm.invoke("推荐一部科幻电影") print(type(result)) # <class 'MovieRecommendation'> print(result.title) # 星际穿越 print(result.year) # 2014 print(result.rating) # 9.4 # 批量推荐（多个 schema） from typing import List as TypingList class MovieList(BaseModel): movies: TypingList[MovieRecommendation] = Field(description="电影推荐列表") structured_llm_list = llm.with_structured_output(MovieList) result = structured_llm_list.invoke("推荐 3 部不同类型的经典电影") for movie in result.movies: print(f"{movie.title}（{movie.year}）- {', '.join(movie.genre)} - {movie.rating}分")
什么时候用 PydanticOutputParser，什么时候用 with_structured_output？
with_structured_output
# 决策树： # 1. 你的模型支持 Function Call 吗？ # - 支持（GPT-4、DeepSeek、Qwen、Claude 等） → 用 with_structured_output() # - 不支持（一些本地模型、老版本模型） → 用 PydanticOutputParser # 2. 你需要在 LCEL 链中使用吗？ # - 需要 → 两者都支持，with_structured_output 更简洁 # - 不需要，只是单次调用 → with_structured_output 更方便 # 3. 你需要自定义验证逻辑吗？ # - 需要复杂验证 → 用 PydanticOutputParser + field_validator # - 不需要 → with_structured_output 足够 # 总结：默认用 with_structured_output，遇到不支持的模型再切换到 PydanticOutputParser
当你只需要让模型输出一个逗号分隔的列表时，这个 Parser 非常方便：
from langchain.output_parsers import CommaSeparatedListOutputParser from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI parser = CommaSeparatedListOutputParser() print(parser.get_format_instructions()) # Your response should be a list of comma separated values, eg: `foo, bar, baz` chain = ( ChatPromptTemplate.from_messages([ ("system", "{format_instructions}"), ("human", "列举 5 种常见的 Python Web 框架"), ]).partial(format_instructions=parser.get_format_instructions()) | ChatOpenAI(model="gpt-4o") | parser ) result = chain.invoke({}) print(result) # ['Django', 'Flask', 'FastAPI', 'Tornado', 'Sanic'] print(type(result)) # <class 'list'>
当输出只能是固定几个值之一时：
from langchain.output_parsers import EnumOutputParser from enum import Enum class Sentiment(str, Enum): POSITIVE = "正面" NEGATIVE = "负面" NEUTRAL = "中性" parser = EnumOutputParser(enum=Sentiment) print(parser.get_format_instructions()) # Select one of the following options: 正面, 负面, 中性 from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI chain = ( ChatPromptTemplate.from_messages([ ("system", "判断情感倾向。{format_instructions}"), ("human", "{text}"), ]).partial(format_instructions=parser.get_format_instructions()) | ChatOpenAI(model="gpt-4o") | parser ) result = chain.invoke({"text": "这款产品太棒了，强烈推荐！"}) print(result) # Sentiment.POSITIVE print(result.value) # "正面" print(result == Sentiment.POSITIVE) # True
from langchain.output_parsers import DatetimeOutputParser parser = DatetimeOutputParser() print(parser.get_format_instructions()) # Write a datetime string that matches the following pattern: '%Y-%m-%dT%H:%M:%S.%fZ' from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI chain = ( ChatPromptTemplate.from_messages([ ("system", "{format_instructions}"), ("human", "{event} 是什么时候？"), ]).partial(format_instructions=parser.get_format_instructions()) | ChatOpenAI(model="gpt-4o") | parser ) result = chain.invoke({"event": "2024年巴黎奥运会开幕式"}) print(type(result)) # <class 'datetime.datetime'> print(result.year) # 2024 print(result.month) # 7
当你不想引入 Pydantic，但又需要多字段结构化输出时：
from langchain.output_parsers import StructuredOutputParser, ResponseSchema # 定义期望的字段 response_schemas = [ ResponseSchema(name="answer", description="对用户问题的回答"), ResponseSchema(name="confidence", description="置信度，0.0 到 1.0 之间的小数"), ResponseSchema(name="sources", description="回答依据的来源，逗号分隔"), ] parser = StructuredOutputParser.from_response_schemas(response_schemas) chain = ( ChatPromptTemplate.from_messages([ ("system", "你是一个知识问答助手。\n\n{format_instructions}"), ("human", "{question}"), ]).partial(format_instructions=parser.get_format_instructions()) | ChatOpenAI(model="gpt-4o") | parser ) result = chain.invoke({"question": "Python 的 GIL 是什么？"}) print(type(result)) # <class 'dict'> print(result["answer"]) # GIL（全局解释器锁）是... print(result["confidence"]) # 0.95 print(result["sources"]) # Python 官方文档, PEP 703
无论你的提示词写得多好，模型偶尔都会输出不符合要求的格式。生产环境必须有完善的解析失败处理策略。
当解析失败时，用另一个 LLM 调用来修复格式问题：
from langchain.output_parsers import OutputFixingParser from langchain_core.output_parsers import PydanticOutputParser from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field class UserInfo(BaseModel): name: str age: int email: str = Field(description="邮箱地址") # 基础 Parser base_parser = PydanticOutputParser(pydantic_object=UserInfo) # 包装成自动修复 Parser fixing_parser = OutputFixingParser.from_llm( parser=base_parser, llm=ChatOpenAI(model="gpt-4o"), # 用于修复的 LLM ) # 模拟一个格式不合法的输出 bad_output = """ 这是用户信息： 姓名：张三 年龄：二十八岁 邮箱：zhangsan@example.com """ # 直接用 base_parser 会失败 try: base_parser.parse(bad_output) except Exception as e: print(f"base_parser 解析失败：{e}") # fixing_parser 会自动发起第二次 LLM 调用来修复格式 result = fixing_parser.parse(bad_output) print(type(result)) # <class 'UserInfo'> print(result.name) # 张三 print(result.age) # 28 print(result.email) # zhangsan@example.com
OutputFixingParser 的工作原理： 1. 先用 base_parser 解析模型输出 2. 如果解析失败，把原始输出、错误信息、格式要求一起发给修复 LLM 3. 修复 LLM 输出符合格式要求的版本 4. 再次用 base_parser 解析修复后的输出
OutputFixingParser
base_parser
适用场景：模型输出质量不稳定、格式错误率较高（>5%）的情况。代价是每次解析失败多一次 LLM 调用。
比 OutputFixingParser 更激进——直接带着原始输入重新请求模型：
from langchain.output_parsers import RetryOutputParser from langchain_core.output_parsers import PydanticOutputParser from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate base_parser = PydanticOutputParser(pydantic_object=UserInfo) retry_parser = RetryOutputParser.from_llm( parser=base_parser, llm=ChatOpenAI(model="gpt-4o"), max_retries=3, # 最多重试 3 次 ) # RetryOutputParser 需要原始 prompt 和 completion prompt = ChatPromptTemplate.from_messages([ ("system", "请输出用户信息。\n{format_instructions}"), ("human", "{query}"), ]).partial(format_instructions=base_parser.get_format_instructions()) # 在链中使用（比 OutputFixingParser 略复杂） from langchain_core.runnables import RunnableParallel, RunnablePassthrough completion_chain = prompt | ChatOpenAI(model="gpt-4o") main_chain = ( RunnableParallel( completion=completion_chain, prompt_value=prompt, ) | RunnableLambda(lambda x: retry_parser.parse_with_prompt( completion=x["completion"].content, prompt_value=x["prompt_value"], )) )
OutputFixingParser vs RetryOutputParser：如何选择
RetryOutputParser
with_retry
实际项目中，最实用的方式是自己在链中处理异常：
from langchain_core.runnables import RunnableLambda from langchain_core.output_parsers import PydanticOutputParser from pydantic import BaseModel, ValidationError from typing import Union import json class AnalysisResult(BaseModel): sentiment: str score: float keywords: list[str] parser = PydanticOutputParser(pydantic_object=AnalysisResult) def safe_parse(ai_message_content: str) -> Union[AnalysisResult, dict]: """ 安全解析：失败时返回包含错误信息的 dict，而不是抛出异常 """ try: return parser.parse(ai_message_content) except (json.JSONDecodeError, ValidationError) as e: # 记录日志（生产环境应发送到日志系统） print(f"[WARN] 解析失败，原始输出：{ai_message_content[:100]}") print(f"[WARN] 错误原因：{e}") # 返回有意义的降级结果 return { "success": False, "error": str(e), "raw_output": ai_message_content, "fallback": { "sentiment": "unknown", "score": 0.5, "keywords": [], } } from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser chain = ( ChatPromptTemplate.from_messages([ ("system", f"分析情感。\n\n{parser.get_format_instructions()}"), ("human", "{text}"), ]) | ChatOpenAI(model="gpt-4o") | StrOutputParser() | RunnableLambda(safe_parse) ) result = chain.invoke({"text": "这个产品还不错"}) # 检查解析是否成功 if isinstance(result, AnalysisResult): print(f"✅ 解析成功：{result.sentiment}，评分 {result.score}") else: print(f"❌ 解析失败，使用降级结果：{result['fallback']}")
当内置 Parser 不满足需求时，你可以继承 BaseOutputParser 实现自定义解析逻辑。
BaseOutputParser
from langchain_core.output_parsers import BaseOutputParser from typing import List import re class NumberListParser(BaseOutputParser[List[int]]): """ 解析模型输出中的所有数字，返回整数列表 适用场景：让模型输出一组数字，不依赖特定格式 """ def parse(self, text: str) -> List[int]: """ 核心解析逻辑 """ # 从文本中提取所有数字 numbers = re.findall(r'\b\d+\b', text) if not numbers: raise ValueError(f"在输出中未找到任何数字：{text}") return [int(n) for n in numbers] def get_format_instructions(self) -> str: """ 告诉模型如何格式化输出（会被注入到 Prompt 里） """ return "请以数字列表的形式输出，每个数字用空格或逗号分隔，例如：1, 2, 3, 4, 5" @property def _type(self) -> str: """Parser 的唯一类型标识""" return "number_list" # 使用 from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI parser = NumberListParser() chain = ( ChatPromptTemplate.from_messages([ ("system", parser.get_format_instructions()), ("human", "{question}"), ]) | ChatOpenAI(model="gpt-4o") | parser ) result = chain.invoke({"question": "给我 5 个质数"}) print(result) # [2, 3, 5, 7, 11] print(type(result)) # <class 'list'>
一个更实用的例子：解析模型输出的 Markdown 表格：
from langchain_core.output_parsers import BaseOutputParser from typing import List, Dict import re class MarkdownTableParser(BaseOutputParser[List[Dict[str, str]]]): """ 解析模型输出的 Markdown 表格，返回字典列表 输入： | 姓名 | 年龄 | 城市 | |------|------|------| | 张三 | 28 | 北京 | | 李四 | 32 | 上海 | 输出： [ {"姓名": "张三", "年龄": "28", "城市": "北京"}, {"姓名": "李四", "年龄": "32", "城市": "上海"}, ] """ def parse(self, text: str) -> List[Dict[str, str]]: lines = [line.strip() for line in text.strip().split('\n')] # 过滤掉空行和分隔线 table_lines = [ line for line in lines if line.startswith('|') and not re.match(r'^\|[\s\-\|]+\|$', line) ] if len(table_lines) < 2: raise ValueError(f"未找到有效的 Markdown 表格，原始输出：\n{text}") # 解析表头 headers = [cell.strip() for cell in table_lines[0].split('|') if cell.strip()] # 解析数据行 rows = [] for line in table_lines[1:]: cells = [cell.strip() for cell in line.split('|') if cell.strip()] if len(cells) == len(headers): rows.append(dict(zip(headers, cells))) return rows def get_format_instructions(self) -> str: return ( "请以 Markdown 表格格式输出，包含表头行和数据行，" "用 | 分隔列，第二行为 |---|---|...| 格式的分隔线。" ) @property def _type(self) -> str: return "markdown_table" # 使用 from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI parser = MarkdownTableParser() chain = ( ChatPromptTemplate.from_messages([ ("system", parser.get_format_instructions()), ("human", "{request}"), ]) | ChatOpenAI(model="gpt-4o") | parser ) result = chain.invoke({"request": "列出 5 个世界著名城市，包含城市名、国家、人口（万）三列"}) for row in result: print(row) # {'城市名': '东京', '国家': '日本', '人口（万）': '1400'} # {'城市名': '纽约', '国家': '美国', '人口（万）': '840'} # ...
如果你的自定义 Parser 需要支持流式解析，要继承 BaseTransformOutputParser：
BaseTransformOutputParser
from langchain_core.output_parsers import BaseTransformOutputParser from typing import Iterator, AsyncIterator class StreamingUpperCaseParser(BaseTransformOutputParser[str]): """ 流式大写转换 Parser：在流式输出时把每个 chunk 转为大写 继承 BaseTransformOutputParser 以支持流式 """ def parse(self, text: str) -> str: """同步版本""" return text.upper() def transform(self, input: Iterator, config=None, **kwargs) -> Iterator[str]: """流式版本：处理每个输入 chunk""" for chunk in input: if hasattr(chunk, 'content'): # AIMessageChunk yield chunk.content.upper() else: yield str(chunk).upper() @property def _type(self) -> str: return "streaming_uppercase" # 测试流式支持 from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate parser = StreamingUpperCaseParser() chain = ChatPromptTemplate.from_messages([("human", "{q}")]) | ChatOpenAI(model="gpt-4o") | parser # 同步调用 result = chain.invoke({"q": "说 hello world"}) print(result) # HELLO WORLD（大写） # 流式调用（每个 chunk 实时转大写） for chunk in chain.stream({"q": "慢慢说 hello"}): print(chunk, end="", flush=True)
把本篇所有知识点综合应用，构建一个从非结构化文本中提取结构化信息的完整流水线：
import os from typing import List, Optional from dotenv import load_dotenv from pydantic import BaseModel, Field, field_validator from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import PydanticOutputParser from langchain.output_parsers import OutputFixingParser from langchain_core.runnables import RunnableLambda, RunnablePassthrough from langchain_core.output_parsers import StrOutputParser load_dotenv() # ========== 定义数据模型 ========== class ContactInfo(BaseModel): """联系信息""" phone: Optional[str] = Field(None, description="电话号码") email: Optional[str] = Field(None, description="邮箱地址") address: Optional[str] = Field(None, description="地址") class PersonProfile(BaseModel): """从文本中提取的人物信息""" full_name: str = Field(description="全名") age: Optional[int] = Field(None, description="年龄，如果文中未提及则为 null") occupation: Optional[str] = Field(None, description="职业") organization: Optional[str] = Field(None, description="所属组织或公司") skills: List[str] = Field(default_factory=list, description="技能列表") contact: ContactInfo = Field(default_factory=ContactInfo, description="联系方式") bio_summary: str = Field(description="50字以内的人物简介") @field_validator("full_name") @classmethod def name_must_not_be_empty(cls, v: str) -> str: if not v.strip(): raise ValueError("姓名不能为空") return v.strip() @field_validator("age") @classmethod def age_must_be_reasonable(cls, v: Optional[int]) -> Optional[int]: if v is not None and not (0 < v < 150): raise ValueError(f"年龄不合理：{v}") return v # ========== 构建信息抽取链 ========== llm = ChatOpenAI(model="gpt-4o", temperature=0) base_parser = PydanticOutputParser(pydantic_object=PersonProfile) # 使用 OutputFixingParser 提高鲁棒性 robust_parser = OutputFixingParser.from_llm( parser=base_parser, llm=llm, ) # 信息抽取 Prompt extract_prompt = ChatPromptTemplate.from_messages([ ("system", """你是一个专业的信息抽取助手。 请从用户提供的文本中精确抽取人物信息，严格按照以下格式输出： {format_instructions} 重要说明： - 只提取文本中明确出现的信息，不要推断或捏造 - 未提及的字段填写 null - bio_summary 用中文概括"""), ("human", "请从以下文本中提取人物信息：\n\n{text}"), ]).partial(format_instructions=base_parser.get_format_instructions()) # 完整的抽取链 extract_chain = ( extract_prompt | llm | StrOutputParser() | RunnableLambda(lambda x: robust_parser.parse(x)) ) # ========== 批量处理器 ========== class InfoExtractionPipeline: """ 批量信息抽取流水线 """ def __init__(self, chain, max_concurrency: int = 3): self.chain = chain self.max_concurrency = max_concurrency self.results = [] self.errors = [] def process_batch(self, texts: List[str]) -> List[dict]: """批量处理文本""" inputs = [{"text": t} for t in texts] raw_results = self.chain.batch( inputs, config={"max_concurrency": self.max_concurrency}, return_exceptions=True, ) processed = [] for i, result in enumerate(raw_results): if isinstance(result, Exception): self.errors.append({"index": i, "text": texts[i][:50], "error": str(result)}) processed.append(None) else: self.results.append(result) processed.append(result) return processed def report(self): """输出处理报告""" total = len(self.results) + len(self.errors) print(f"\n=== 处理报告 ===") print(f"总计：{total} 条") print(f"成功：{len(self.results)} 条") print(f"失败：{len(self.errors)} 条") if self.errors: print("\n失败详情：") for err in self.errors: print(f" [{err['index']}] {err['text']}... → {err['error']}") # ========== 使用示例 ========== test_texts = [ """ 王芳，女，35岁，现任阿里巴巴集团高级产品经理。 毕业于复旦大学计算机系，有10年互联网产品经验。 擅长用户研究、产品规划和数据分析。 联系邮箱：wangfang@alibaba.com，电话：13812345678。 """, """ 李明是一名自由软件工程师，精通 Python、Go 和 Kubernetes。 曾在腾讯工作 5 年，现在专注于开源项目贡献。 GitHub: github.com/liming，常驻深圳南山区。 """, """ 张博士，42岁，清华大学人工智能研究所研究员。 主要研究方向为自然语言处理和机器学习，发表 SCI 论文 50 余篇。 邮箱：zhang@tsinghua.edu.cn """, ] pipeline = InfoExtractionPipeline(extract_chain) results = pipeline.process_batch(test_texts) for i, profile in enumerate(results): if profile: print(f"\n--- 人物 {i+1} ---") print(f"姓名：{profile.full_name}") print(f"年龄：{profile.age}") print(f"职业：{profile.occupation}") print(f"组织：{profile.organization}") print(f"技能：{', '.join(profile.skills)}") print(f"简介：{profile.bio_summary}") if profile.contact.email: print(f"邮箱：{profile.contact.email}") pipeline.report()
OutputParser 不只是"链末尾的一个组件"，它还有很多值得深入了解的使用模式，能让你的链更灵活、更健壮。
Parser 不一定非得放在链末尾。在某些场景下，你需要在链的中间解析一步的输出，然后基于解析结果决定后续逻辑：
from langchain_core.output_parsers import PydanticOutputParser, StrOutputParser from langchain_core.runnables import RunnablePassthrough, RunnableLambda from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from pydantic import BaseModel, Field class IntentClassification(BaseModel): intent: str = Field(description="用户意图：question / task / chat") confidence: float = Field(description="置信度 0-1") language: str = Field(description="用户使用的语言：zh / en / other") intent_parser = PydanticOutputParser(pydantic_object=IntentClassification) # 第一步：分类意图 classify_chain = ( ChatPromptTemplate.from_messages([ ("system", f"分析用户输入的意图。\n{intent_parser.get_format_instructions()}"), ("human", "{user_input}"), ]) | ChatOpenAI(model="gpt-4o", temperature=0) | StrOutputParser() | RunnableLambda(intent_parser.parse) # 在链中间解析 ) # 第二步：根据意图选择不同的响应策略 def build_response_chain(intent_result: IntentClassification): """基于意图分类结果选择响应链""" lang_instruction = "请用中文回答" if intent_result.language == "zh" else "Please respond in English" if intent_result.intent == "question": prompt = ChatPromptTemplate.from_messages([ ("system", f"你是知识问答助手，给出准确详细的回答。{lang_instruction}"), ("human", "{user_input}"), ]) elif intent_result.intent == "task": prompt = ChatPromptTemplate.from_messages([ ("system", f"你是任务执行助手，给出具体可操作的步骤。{lang_instruction}"), ("human", "{user_input}"), ]) else: # chat prompt = ChatPromptTemplate.from_messages([ ("system", f"你是友好的聊天伙伴，轻松自然地对话。{lang_instruction}"), ("human", "{user_input}"), ]) return prompt | ChatOpenAI(model="gpt-4o") | StrOutputParser() # 组合：先分类，再基于分类结果路由 full_chain = ( RunnablePassthrough.assign(intent=classify_chain) | RunnableLambda(lambda x: build_response_chain(x["intent"]).invoke(x)) ) result = full_chain.invoke({"user_input": "Python 的 GIL 是什么？"}) print(result)
有时候模型的输出是"JSON 里包含 Markdown 格式的文本"，需要多层解析：
from langchain_core.output_parsers import JsonOutputParser, BaseOutputParser from pydantic import BaseModel from typing import List import markdown from bs4 import BeautifulSoup class ArticleSection(BaseModel): title: str content: str word_count: int class Article(BaseModel): title: str sections: List[ArticleSection] total_words: int # 第一层：解析 JSON 结构 json_parser = JsonOutputParser() # 第二层：验证并转换成 Pydantic 对象 def validate_article(data: dict) -> Article: sections = [] total = 0 for s in data.get("sections", []): wc = len(s["content"].split()) sections.append(ArticleSection( title=s["title"], content=s["content"], word_count=wc, )) total += wc return Article( title=data["title"], sections=sections, total_words=total, ) from langchain_core.runnables import RunnableLambda from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate chain = ( ChatPromptTemplate.from_messages([ ("system", '以 JSON 格式生成文章大纲，结构：{"title": "...", "sections": [{"title": "...", "content": "..."}]}'), ("human", "写一篇关于 {topic} 的文章大纲"), ]) | ChatOpenAI(model="gpt-4o") | json_parser # 第一层：JSON 解析 | RunnableLambda(validate_article) # 第二层：Pydantic 验证 ) result = chain.invoke({"topic": "Python 异步编程"}) print(type(result)) # <class 'Article'> print(f"文章标题：{result.title}") print(f"总字数：{result.total_words}") for s in result.sections: print(f" - {s.title}（{s.word_count} 词）")
不同 Parser 对流式输出的支持程度不同，这是开发者常遇到的困惑：
from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser, JsonOutputParser llm = ChatOpenAI(model="gpt-4o") prompt = ChatPromptTemplate.from_messages([("human", "{q}")]) # ✅ StrOutputParser：完美支持流式，每个 chunk 立即输出 chain_str = prompt | llm | StrOutputParser() for chunk in chain_str.stream({"q": "介绍 Python"}): print(chunk, end="", flush=True) # 实时输出每个字 print("\n\n---\n") # ✅ JsonOutputParser：支持流式，但输出的是增量解析的字典 chain_json = ( ChatPromptTemplate.from_messages([ ("system", "只输出 JSON"), ("human", '生成 3 个编程语言信息，格式：{{"languages": [...]}}'), ]) | llm | JsonOutputParser() ) for chunk in chain_json.stream({}): # chunk 是每次解析到的部分字典，随着 token 流入逐渐完整 print(f"\r当前状态：{str(chunk)[:80]}", end="", flush=True) print() # ⚠️ PydanticOutputParser：不支持流式解析 # 只有在完整输出后才能解析成 Pydantic 对象 # 如果需要"流式 + Pydantic"，先用 JsonOutputParser 流式，最后用 Pydantic 验证 from pydantic import BaseModel class LangInfo(BaseModel): name: str year: int def stream_then_validate(chain, query): """先流式收集，再 Pydantic 验证""" import json full_output = "" for chunk in chain.stream(query): # 这里只是演示思路，实际用 JsonOutputParser 流式更优雅 if isinstance(chunk, str): full_output += chunk else: full_output = json.dumps(chunk) try: return LangInfo(**json.loads(full_output)) except Exception as e: print(f"验证失败：{e}") return None
流式支持情况总结：
StrOutputParser
CommaSeparatedListOutputParser
有时候格式指令本身也需要动态生成，比如字段描述要根据用户语言变化：
from langchain_core.output_parsers import PydanticOutputParser from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnableLambda, RunnablePassthrough from langchain_openai import ChatOpenAI from pydantic import BaseModel from langchain_core.output_parsers import StrOutputParser class Summary(BaseModel): title: str key_points: list[str] conclusion: str def create_chain_for_language(language: str): """根据目标语言创建不同的解析链""" # 动态修改字段描述（虽然 Pydantic Parser 本身不支持动态描述， # 但可以在 system prompt 里用语言声明来引导） parser = PydanticOutputParser(pydantic_object=Summary) lang_map = { "zh": "请使用中文填写所有字段", "en": "Please fill all fields in English", "ja": "すべてのフィールドを日本語で記入してください", } lang_instruction = lang_map.get(language, "Use the same language as the input") return ( ChatPromptTemplate.from_messages([ ("system", f"""你是一个文档摘要助手。{lang_instruction} {parser.get_format_instructions()}"""), ("human", "请摘要以下内容：\n\n{text}"), ]) | ChatOpenAI(model="gpt-4o") | StrOutputParser() | RunnableLambda(parser.parse) ) # 中文摘要 zh_chain = create_chain_for_language("zh") result = zh_chain.invoke({"text": "LangChain is a framework for building LLM applications..."}) print(result.title) # 中文标题 print(result.key_points) # 中文要点列表
根据你的具体需求快速选择合适的 Parser：
我需要... │ ├─ 纯文本输出 │ └─ StrOutputParser ✓ │ ├─ 结构化数据输出 │ ├─ 模型支持 Function Call（GPT-4、DeepSeek、Claude 等） │ │ └─ with_structured_output(PydanticModel) ✓ 【最推荐】 │ │ │ ├─ 模型不支持 Function Call │ │ ├─ 需要 Pydantic 对象（有类型验证） │ │ │ └─ PydanticOutputParser ✓ │ │ ├─ 只需要 dict（无需验证） │ │ │ └─ JsonOutputParser ✓ │ │ └─ 轻量级多字段，不用 Pydantic │ │ └─ StructuredOutputParser ✓ │ │ │ ├─ 简单列表 │ │ └─ CommaSeparatedListOutputParser ✓ │ │ │ ├─ 枚举值之一 │ │ └─ EnumOutputParser ✓ │ │ │ └─ 日期时间 │ └─ DatetimeOutputParser ✓ │ ├─ 解析失败率较高，需要自动修复 │ ├─ 内容正确，格式有问题 → OutputFixingParser ✓ │ └─ 内容也不对，需要重新生成 → RetryOutputParser ✓ │ └─ 以上都不满足（特殊格式、特殊逻辑） └─ 自定义 BaseOutputParser ✓
parse
get_format_instructions
你现在应该能做到： - 根据具体场景选择最合适的 Parser，不再手写 JSON 解析逻辑 - 用 PydanticOutputParser + field_validator 构建带验证的结构化输出 - 用 with_structured_output 一行代码实现类型安全的结构化输出 - 用 OutputFixingParser 包装现有 Parser 提升解析成功率 - 实现自定义 Parser 处理特殊格式的模型输出 - 构建批量信息抽取流水线，包含错误处理和处理报告
field_validator
《记忆模块——给对话加上上下文管理》
解决了输入（Prompt）和输出（Parser）之后，下一篇进入多轮对话的核心——Memory。你会学到：
RunnableWithMessageHistory
trim_messages
filter_messages
代码仓库：本系列所有可运行代码示例统一维护在 GitHub，每篇对应独立目录，可直接克隆运行。 系列导航：[第一篇] → [第二篇] → [第三篇] → [第四篇] → 第五篇（当前） → 第六篇 → ...
代码仓库：本系列所有可运行代码示例统一维护在 GitHub，每篇对应独立目录，可直接克隆运行。
系列导航：[第一篇] → [第二篇] → [第三篇] → [第四篇] → 第五篇（当前） → 第六篇 → ...
还没有评论，来抢沙发吧！
博客管理员
40 篇文章
admin
还没有评论，来抢沙发吧！