LangChain學習——結構化輸出和數據解析

LangChain

本指南全面介紹LangChain中結構化輸出生成和數據解析的核心功能，包括Pydantic BaseModel構造、各種輸出解析器的使用，以及高級錯誤處理機制。

詳細測試樣例和代碼可參考如下兩個鏈接：

test_output_parsers
test_pydantic_base_model

結構化輸出概述

概念定義

結構化輸出是指將大語言模型（LLM）的自然語言輸出轉換為具有明確數據類型和格式的結構化數據。這種轉換使得AI應用能夠可靠地處理和使用LLM的輸出結果。

核心優勢

🎯 類型安全：確保數據符合預定義的結構和類型約束
🔄 一致性：保證輸出格式的穩定性和可預測性
? 可靠性：通過驗證和錯誤處理提高系統魯棒性
🔧 易集成：直接獲得可在應用中使用的數據對象

技術實現方式

LangChain提供兩種主要的結構化輸出實現方式：

結構化輸出模式（Structured Output）：基于約束解碼技術，確保100%格式正確
函數調用模式（Function Calling）：基于工具調用機制，支持更復雜的數據結構

相關鏈接：

結構化輸出概念
輸出解析器概念

Pydantic BaseModel 集成

概念定義

Pydantic BaseModel是Python中最強大的數據驗證和序列化庫，LangChain深度集成了Pydantic來定義和驗證結構化輸出的數據模型。

with_structured_output 方法

輸入：Pydantic模型類或JSON Schema
輸出：符合指定結構的數據對象
原理：將LLM輸出約束到預定義的數據結構中

from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAIclass UserProfile(BaseModel):"""用戶檔案數據模型"""name: str = Field(description="用戶姓名")age: int = Field(description="用戶年齡", ge=0, le=150)email: str = Field(description="郵箱地址")skills: List[str] = Field(description="技能列表")# 創建結構化輸出模型
model = ChatOpenAI(model="gpt-4o-mini")
structured_llm = model.with_structured_output(UserProfile)# 使用結構化輸出
result = structured_llm.invoke("提取用戶信息：張三，25歲，郵箱zhang@example.com，擅長Python和數據分析")
# result 是 UserProfile 類型的對象
print(f"姓名: {result.name}, 年齡: {result.age}")

復雜數據結構支持

嵌套模型：支持模型之間的嵌套關系

class Address(BaseModel):"""地址信息"""street: str = Field(description="街道地址")city: str = Field(description="城市")country: str = Field(description="國家")class Company(BaseModel):"""公司信息"""name: str = Field(description="公司名稱")address: Address = Field(description="公司地址")employees: List[UserProfile] = Field(description="員工列表")# 處理復雜嵌套結構
company_llm = model.with_structured_output(Company)

枚舉類型支持

from enum import Enumclass TaskStatus(str, Enum):"""任務狀態枚舉"""PENDING = "pending"IN_PROGRESS = "in_progress"COMPLETED = "completed"CANCELLED = "cancelled"class Task(BaseModel):"""任務信息"""title: str = Field(description="任務標題")status: TaskStatus = Field(description="任務狀態")priority: int = Field(description="優先級", ge=1, le=5)task_llm = model.with_structured_output(Task)

Function Calling vs Structured Output

技術選擇指南：

特性	Structured Output	Function Calling
動態字段支持	? 不支持 `Dict[str, Any]`	? 完全支持
復雜嵌套	受限	無限制
格式可靠性	100%正確	高度可靠
首次延遲	有預處理開銷	無額外開銷

# 包含動態字段的模型需要使用 function_calling
class FlexibleResponse(BaseModel):core_data: str = Field(description="核心數據")metadata: Dict[str, Any] = Field(description="動態元數據")# 必須指定 method="function_calling"
flexible_llm = model.with_structured_output(FlexibleResponse,method="function_calling"
)

相關鏈接：

如何返回結構化數據
Pydantic模型驗證

輸出解析器系統

概念定義

**輸出解析器（Output Parsers）**是LangChain中負責將LLM的文本輸出轉換為特定數據類型的組件。它們提供了比結構化輸出更靈活的數據處理能力。

基礎解析器類型

StrOutputParser

功能：提取AI消息中的純文本內容
輸入：AIMessage對象
輸出：字符串
適用場景：簡單的文本提取和處理

from langchain_core.output_parsers import StrOutputParser# 文本提取鏈
text_chain = model | StrOutputParser()
result = text_chain.invoke("介紹一下人工智能")
# result 是純字符串

JsonOutputParser

功能：解析JSON格式的模型輸出
輸入：包含JSON的文本
輸出：Python字典或列表
適用場景：結構化數據提取

from langchain_core.output_parsers import JsonOutputParserjson_parser = JsonOutputParser()# 創建JSON解析鏈
json_chain = model | json_parser
result = json_chain.invoke("用JSON格式返回用戶信息：姓名張三，年齡25")
# result 是 Python 字典

XMLOutputParser

功能：解析XML格式的結構化數據
輸入：XML格式文本
輸出：解析后的數據結構
適用場景：處理層次化數據

from langchain_core.output_parsers import XMLOutputParserxml_parser = XMLOutputParser()
xml_chain = model | xml_parser

YAMLOutputParser

功能：解析YAML格式的配置數據
輸入：YAML格式文本
輸出：Python數據結構
適用場景：配置文件處理

from langchain_core.output_parsers import YamlOutputParseryaml_parser = YamlOutputParser()
yaml_chain = model | yaml_parser

PydanticOutputParser

功能：將文本輸出解析為Pydantic模型實例
輸入：符合模型格式的文本
輸出：Pydantic模型對象
原理：結合格式指令和數據驗證

from langchain_core.output_parsers import PydanticOutputParser# 創建解析器
parser = PydanticOutputParser(pydantic_object=UserProfile)# 獲取格式指令
format_instructions = parser.get_format_instructions()# 構建提示模板
from langchain_core.prompts import ChatPromptTemplateprompt = ChatPromptTemplate.from_template("提取用戶信息：{query}\n{format_instructions}"
).partial(format_instructions=format_instructions)# 創建完整鏈
chain = prompt | model | parser
result = chain.invoke({"query": "張三，工程師，28歲"})
# result 是 UserProfile 對象

流式解析支持

概念：支持實時處理模型的流式輸出
優勢：提供更好的用戶體驗和響應性

# 流式JSON解析
from langchain_core.output_parsers import SimpleJsonOutputParserstreaming_parser = SimpleJsonOutputParser()# 流式處理
for chunk in (model | streaming_parser).stream("生成用戶數據"):print(chunk, end="", flush=True)

相關鏈接：

如何解析字符串輸出
如何解析JSON輸出
如何解析XML輸出
如何解析YAML輸出

智能錯誤處理

概念定義

智能錯誤處理是LangChain提供的高級功能，能夠自動檢測、修復和重試解析錯誤，大幅提升系統的魯棒性和可靠性。

OutputFixingParser

功能：自動修復格式錯誤的LLM輸出
輸入：可能包含格式錯誤的文本
輸出：修復后的正確數據結構
原理：使用另一個LLM理解并修復格式問題

from langchain_core.output_parsers import OutputFixingParser# 基礎解析器
base_parser = JsonOutputParser()# 包裝為自動修復解析器
fixing_parser = OutputFixingParser.from_llm(parser=base_parser,llm=ChatOpenAI(temperature=0.1)  # 使用低溫度提高修復準確性
)# 自動處理格式錯誤
broken_json = '{"name": "張三", "age": 25'  # 缺少閉合括號
fixed_result = fixing_parser.parse(broken_json)  # 自動修復成功

應用場景：

JSON格式錯誤（缺少括號、多余逗號）
XML標簽不匹配
YAML縮進問題
Pydantic字段類型錯誤

RetryWithErrorOutputParser

功能：解析失敗時的智能重試機制
輸入：原始提示和錯誤信息
輸出：重新生成的正確格式數據
原理：將錯誤反饋給LLM，指導重新生成

from langchain_core.output_parsers import RetryWithErrorOutputParser# 創建重試解析器
retry_parser = RetryWithErrorOutputParser.from_llm(parser=PydanticOutputParser(pydantic_object=UserProfile),llm=ChatOpenAI(temperature=0.1),max_retries=3  # 最多重試3次
)# 自動處理解析失敗并重試
chain = prompt | model | retry_parser
result = chain.invoke({"query": "用戶信息提取"})

重試流程：

初始解析嘗試失敗
將錯誤信息反饋給LLM
LLM重新生成符合格式的輸出
重復直到成功或達到最大重試次數

組合錯誤處理策略

組合使用：將多種錯誤處理機制結合，實現最高可靠性

# 構建三層錯誤處理
base_parser = PydanticOutputParser(pydantic_object=ComplexModel)# 第一層：格式修復
fixing_parser = OutputFixingParser.from_llm(parser=base_parser,llm=ChatOpenAI(temperature=0.1)
)# 第二層：重試機制
ultimate_parser = RetryWithErrorOutputParser.from_llm(parser=fixing_parser,llm=ChatOpenAI(temperature=0.1),max_retries=2
)ultimate_chain = prompt | model | ultimate_parser

相關鏈接：

如何處理解析重試
如何修復解析錯誤

高級應用模式

自定義解析器開發

概念：基于BaseOutputParser創建專用的解析邏輯
原理：繼承基類并實現parse和get_format_instructions方法

from langchain_core.output_parsers import BaseOutputParser
from typing import Listclass ListOutputParser(BaseOutputParser[List[str]]):"""列表解析器 - 分隔符解析"""def __init__(self, separator: str = ","):self.separator = separatordef parse(self, text: str) -> List[str]:"""解析分隔符分隔的列表"""return [item.strip() for item in text.split(self.separator)]def get_format_instructions(self) -> str:"""返回格式指令"""return f"請用{self.separator}分隔列表項目"# 使用自定義解析器
list_parser = ListOutputParser(separator="|")
list_chain = model | list_parser

條件解析器

功能：根據內容特征自動選擇解析策略

class ConditionalOutputParser(BaseOutputParser):"""條件解析器 - 智能格式識別"""def parse(self, text: str) -> Union[str, Dict, List]:"""根據內容特征選擇解析策略"""text = text.strip()if text.startswith('{') and text.endswith('}'):# JSON對象格式return json.loads(text)elif text.startswith('[') and text.endswith(']'):# JSON數組格式  return json.loads(text)elif ',' in text:# 逗號分隔的列表return [item.strip() for item in text.split(',')]else:# 純文本return textdef get_format_instructions(self) -> str:return "可以返回JSON、列表或純文本格式"

鏈式解析器

概念：組合多個解析器，實現復雜的數據處理流程

class ChainedOutputParser(BaseOutputParser[Dict[str, Any]]):"""鏈式解析器 - 多策略組合"""def __init__(self, parsers: List[Tuple[str, BaseOutputParser]]):self.parsers = parsersdef parse(self, text: str) -> Dict[str, Any]:"""依次應用多個解析器"""results = {"original": text}for name, parser in self.parsers:try:results[name] = parser.parse(text)except Exception as e:results[f"{name}_error"] = str(e)return results# 創建鏈式解析器
chained_parser = ChainedOutputParser([("json", JsonOutputParser()),("list", ListOutputParser()),("yaml", YamlOutputParser())
])

驗證解析器

功能：在解析后進行額外的數據驗證和清理

class ValidatingOutputParser(BaseOutputParser[Dict]):"""驗證解析器 - 數據質量保證"""def __init__(self, base_parser: BaseOutputParser, validators: List[callable]):self.base_parser = base_parserself.validators = validatorsdef parse(self, text: str) -> Dict:"""解析并驗證數據"""result = self.base_parser.parse(text)# 應用驗證規則for validator in self.validators:result = validator(result)return resultdef validate_user_age(data: Dict) -> Dict:"""驗證用戶年齡合理性"""if 'age' in data and not (0 <= data['age'] <= 150):data['age'] = max(0, min(data['age'], 150))return data# 使用驗證解析器
validating_parser = ValidatingOutputParser(base_parser=JsonOutputParser(),validators=[validate_user_age]
)

相關鏈接：

如何創建自定義解析器

最佳實踐指南

選擇合適的方法

決策流程圖：

需要結構化輸出？
├─ 是 → 數據結構復雜（包含動態字段）？
│   ├─ 是 → 使用 with_structured_output(method="function_calling")
│   └─ 否 → 使用 with_structured_output() (默認模式)
└─ 否 → 需要格式轉換？├─ 是 → 使用合適的OutputParser (Json/XML/YAML)└─ 否 → 使用 StrOutputParser

錯誤處理最佳實踐

分層錯誤處理：

預防：使用清晰的格式指令
檢測：實現數據驗證邏輯
修復：使用OutputFixingParser
重試：使用RetryWithErrorOutputParser
降級：提供備用解析策略

def robust_structured_output(query: str, model_class: BaseModel, fallback_parser=None):"""魯棒的結構化輸出處理"""try:# 主要方法：結構化輸出structured_llm = model.with_structured_output(model_class)return structured_llm.invoke(query)except Exception as e:if fallback_parser:# 降級方法：使用解析器return fallback_parser.parse(model.invoke(query).content)else:raise e

監控和調試

性能監控：

import time
from typing import Anydef monitored_parse(parser: BaseOutputParser, text: str) -> Dict[str, Any]:"""監控解析性能"""start_time = time.time()try:result = parser.parse(text)success = Trueerror = Noneexcept Exception as e:result = Nonesuccess = Falseerror = str(e)end_time = time.time()return {"result": result,"success": success,"error": error,"parse_time": end_time - start_time,"input_length": len(text)}

提示工程優化

格式指令優化：

def enhance_format_instructions(base_instructions: str, examples: List[str] = None) -> str:"""增強格式指令"""enhanced = base_instructionsif examples:enhanced += "\n\n示例格式:\n"for i, example in enumerate(examples, 1):enhanced += f"{i}. {example}\n"enhanced += "\n注意：嚴格按照上述格式返回，確保數據完整性。"return enhanced

類型安全增強

泛型解析器：

from typing import TypeVar, GenericT = TypeVar('T')class TypedOutputParser(BaseOutputParser[T], Generic[T]):"""類型安全的解析器基類"""def __init__(self, target_type: Type[T]):self.target_type = target_typedef parse(self, text: str) -> T:# 實現類型安全的解析邏輯pass