LangChain-輸出解析器 (Output Parsers)

輸出解析器是LangChain的重要組件，用于將語言模型的原始文本輸出轉換為結構化數據。本文檔詳細介紹了輸出解析器的類型、功能和最佳實踐。

概述

語言模型通常輸出自然語言文本，但在應用開發中，我們經常需要將這些文本轉換為結構化的數據格式，如列表、字典或對象。輸出解析器實現了這一關鍵功能，提供了：

結構化數據提取：將自然語言轉換為特定數據結構
格式一致性保證：確保輸出符合預期格式
容錯處理：處理模型輸出不符合要求的情況
模型指導：向模型提供正確輸出格式的指導

基礎解析器

1. 字符串解析器 (StrOutputParser)

最簡單的解析器，用于提取模型返回的原始文本：

from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate# 創建簡單鏈
model = ChatOpenAI()
prompt = ChatPromptTemplate.from_template("解釋{topic}是什么？")
parser = StrOutputParser()# 組合組件
chain = prompt | model | parser# 調用鏈
result = chain.invoke({"topic": "量子計算"})# result 是一個字符串，包含模型對量子計算的解釋

2. 列表解析器 (CommaSeparatedListOutputParser)

將逗號分隔的文本轉換為Python列表：

from langchain_core.output_parsers import CommaSeparatedListOutputParser# 創建列表解析器
list_parser = CommaSeparatedListOutputParser()# 創建帶格式說明的提示模板
format_instructions = list_parser.get_format_instructions()
list_prompt = ChatPromptTemplate.from_template("列出{topic}的五個最重要特點。\n{format_instructions}"
)# 組合組件
list_chain = list_prompt | model | list_parser# 調用鏈
topic_features = list_chain.invoke({"topic": "人工智能", "format_instructions": format_instructions
})# 結果是一個Python列表: ['特點1', '特點2', '特點3', '特點4', '特點5']

3. JSON解析器 (JsonOutputParser)

將JSON格式文本轉換為Python字典或列表：

from langchain_core.output_parsers import JsonOutputParser# 創建JSON解析器
json_parser = JsonOutputParser()# 創建帶格式說明的提示模板
json_format_instructions = json_parser.get_format_instructions()
json_prompt = ChatPromptTemplate.from_template("生成一個包含{person}基本信息的JSON。應包括姓名、職業、年齡和技能列表。\n{format_instructions}"
)# 組合組件
json_chain = json_prompt | model | json_parser# 調用鏈
person_info = json_chain.invoke({"person": "愛因斯坦","format_instructions": json_format_instructions
})# 結果是一個Python字典，包含愛因斯坦的信息

高級解析器

1. Pydantic解析器 (PydanticOutputParser)

使用Pydantic模型定義輸出結構：

from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field, validator
from typing import List# 定義Pydantic模型
class Movie(BaseModel):title: str = Field(description="電影標題")director: str = Field(description="導演姓名")year: int = Field(description="上映年份")genre: List[str] = Field(description="電影類型")rating: float = Field(description="評分（1-10）")@validator("rating")def rating_must_be_valid(cls, v):if v < 1 or v > 10:raise ValueError("評分必須在1到10之間")return v# 創建Pydantic解析器
pydantic_parser = PydanticOutputParser(pydantic_object=Movie)# 創建帶格式說明的提示模板
format_instructions = pydantic_parser.get_format_instructions()
pydantic_prompt = ChatPromptTemplate.from_template("生成一部{genre}電影的信息。\n{format_instructions}"
)# 組合組件
pydantic_chain = pydantic_prompt | model | pydantic_parser# 調用鏈
movie_data = pydantic_chain.invoke({"genre": "科幻","format_instructions": format_instructions
})# 結果是一個Movie對象，包含電影信息

2. XML解析器 (XMLOutputParser)

解析XML格式的輸出：

from langchain_core.output_parsers import XMLOutputParser# 創建XML解析器
xml_parser = XMLOutputParser()# 創建帶格式說明的提示模板
xml_format = """<analysis><sentiment>positive or negative</sentiment><language>detected language</language><summary>brief summary</summary>
</analysis>"""xml_prompt = ChatPromptTemplate.from_template("分析以下文本的情感、語言和內容：\n{text}\n\n以下面的XML格式輸出:\n{xml_format}"
)# 組合組件
xml_chain = xml_prompt | model | xml_parser# 調用鏈
xml_result = xml_chain.invoke({"text": "人工智能正在迅速發展，為各行各業帶來革命性變化。","xml_format": xml_format
})# 結果是一個包含解析XML數據的字典

3. 自定義解析器

創建自定義輸出解析器：

from langchain_core.output_parsers import BaseOutputParser
from typing import Dict, Anyclass CustomKeyValueParser(BaseOutputParser[Dict[str, Any]]):"""解析形如'key: value'的文本"""def parse(self, text: str) -> Dict[str, Any]:"""從文本中解析鍵值對"""result = {}lines = text.strip().split('\n')for line in lines:if ':' in line:key, value = line.split(':', 1)result[key.strip()] = value.strip()return resultdef get_format_instructions(self) -> str:"""提供格式指導給模型"""return """請以'鍵: 值'的格式返回信息，每行一個鍵值對。
例如：
名稱: 愛因斯坦
職業: 物理學家
貢獻: 相對論"""# 使用自定義解析器
custom_parser = CustomKeyValueParser()custom_prompt = ChatPromptTemplate.from_template("提供關于{person}的基本信息。\n{format_instructions}"
)custom_chain = custom_prompt | model | custom_parserresult = custom_chain.invoke({"person": "牛頓","format_instructions": custom_parser.get_format_instructions()
})

組合解析器

1. 多重解析 (RouterOutputParser)

根據內容選擇不同的解析器：

from langchain_core.output_parsers import RouterOutputParser
from langchain_core.output_parsers.openai_tools import PydanticToolsParserclass Person(BaseModel):name: strage: intclass Company(BaseModel):name: strindustry: stremployees: int# 創建組合解析器
router_parser = PydanticToolsParser(tools=[Person,Company
])# 創建提示模板
router_prompt = ChatPromptTemplate.from_template("根據查詢決定是返回人物信息還是公司信息。\n查詢: {query}"
)# 組合組件
router_chain = router_prompt | model.bind_tools([Person, Company]) | router_parser# 人物查詢
person_result = router_chain.invoke({"query": "告訴我關于比爾蓋茨的信息"})
# 返回Person對象# 公司查詢
company_result = router_chain.invoke({"query": "告訴我關于微軟的信息"})
# 返回Company對象

2. 錯誤處理與重試 (OutputFixingParser)

處理解析錯誤并嘗試修復：

from langchain_core.output_parsers import OutputFixingParser# 創建基礎解析器（容易出錯的格式）
base_parser = PydanticOutputParser(pydantic_object=Movie)# 創建帶修復功能的解析器
fixing_parser = OutputFixingParser.from_llm(parser=base_parser,llm=ChatOpenAI()
)# 即使模型輸出不完全符合格式要求，也能嘗試修復并解析
try_chain = prompt | model | fixing_parser

構建復雜解析策略

1. 結構化提取與轉換

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import PydanticOutputParser, StrOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.runnables import RunnablePassthrough, RunnableBranch
from typing import List, Optional# 定義多級數據結構
class Author(BaseModel):name: str = Field(description="作者姓名")background: Optional[str] = Field(description="作者背景", default=None)class Book(BaseModel):title: str = Field(description="書名")year: int = Field(description="出版年份")author: Author = Field(description="作者信息")genres: List[str] = Field(description="圖書類型")summary: str = Field(description="內容簡介")# 創建解析器和提示
book_parser = PydanticOutputParser(pydantic_object=Book)
instructions = book_parser.get_format_instructions()book_prompt = ChatPromptTemplate.from_template("提供關于{book_title}的詳細信息，包括作者、出版年份、類型和簡介。\n{format_instructions}"
)# 處理結果的函數
def process_book(book: Book) -> dict:# 計算出版至今年數import datetimecurrent_year = datetime.datetime.now().yearyears_since_pub = current_year - book.yearreturn {"book": book,"years_since_publication": years_since_pub,"is_classic": years_since_pub > 50}# 組合處理鏈
book_chain = book_prompt | model | book_parser | process_book# 調用鏈
result = book_chain.invoke({"book_title": "戰爭與和平","format_instructions": instructions
})

2. 條件解析邏輯

from langchain_core.runnables import RunnableBranch
from langchain_core.output_parsers import JsonOutputParser, XMLOutputParser# 創建多格式解析器
def is_json(text):return text.strip().startswith('{')def is_xml(text):return text.strip().startswith('<')# 創建條件分支解析器
multi_format_parser = RunnableBranch((is_json, JsonOutputParser()),(is_xml, XMLOutputParser()),StrOutputParser()  # 默認解析器
)# 在鏈中使用
flexible_chain = prompt | model | multi_format_parser

輸出格式化技巧

1. 明確的格式指令

提供詳細的格式說明，幫助模型生成可解析的輸出：

from langchain_core.output_parsers.json import JsonOutputParserjson_parser = JsonOutputParser()
instructions = json_parser.get_format_instructions()detailed_prompt = """生成一個包含電影信息的JSON對象。必須包含以下字段：
- title: 電影標題 (字符串)
- director: 導演姓名 (字符串)
- year: 上映年份 (整數)
- genres: 電影類型 (字符串數組)
- rating: 評分，1-10之間 (數值){format_instructions}電影: {movie}"""prompt = ChatPromptTemplate.from_template(detailed_prompt)
chain = prompt | model | json_parserresult = chain.invoke({"movie": "黑客帝國", "format_instructions": instructions})

2. 使用Enum限制選項

使用枚舉類型限制可能的值：

from enum import Enum
from langchain_core.pydantic_v1 import BaseModel, Fieldclass Sentiment(str, Enum):POSITIVE = "positive"NEUTRAL = "neutral"NEGATIVE = "negative"class SentimentAnalysis(BaseModel):text: str = Field(description="分析的文本")sentiment: Sentiment = Field(description="文本情感 (positive, neutral, negative)")confidence: float = Field(description="置信度 (0.0-1.0)")class Config:use_enum_values = True# 創建解析器
sentiment_parser = PydanticOutputParser(pydantic_object=SentimentAnalysis)

3. 分步解析復雜輸出

對于復雜任務，先將輸出分解為簡單部分：

from langchain_core.output_parsers import StrOutputParser# 第一步：生成文本分析
analysis_prompt = ChatPromptTemplate.from_template("分析以下文本: {text}")
analysis_chain = analysis_prompt | model | StrOutputParser()# 第二步：從分析中提取結構化數據
structure_prompt = ChatPromptTemplate.from_template("""基于下面的分析，提取主要觀點和情感評分(1-10)，以JSON格式返回:分析: {analysis}JSON格式:{{"main_points": ["觀點1", "觀點2", ...],"sentiment_score": 評分}}"""
)
structure_chain = structure_prompt | model | JsonOutputParser()# 組合兩個步驟
full_chain = {"analysis": analysis_chain} | structure_chainresult = full_chain.invoke({"text": "這個產品質量很好，但價格有點貴。客服態度也不錯，總體來說是不錯的購物體驗。"})

解析器驗證與錯誤處理

1. 使用驗證邏輯

Pydantic模型中添加驗證器：

from langchain_core.pydantic_v1 import BaseModel, Field, validator
from typing import Listclass Product(BaseModel):name: str = Field(description="產品名稱")price: float = Field(description="產品價格")features: List[str] = Field(description="產品特點列表")rating: int = Field(description="產品評分(1-5)")@validator("price")def price_must_be_positive(cls, v):if v <= 0:raise ValueError("價格必須為正數")return v@validator("rating")def rating_must_be_valid(cls, v):if v < 1 or v > 5:raise ValueError("評分必須在1到5之間")return v

2. 錯誤處理策略

from langchain_core.output_parsers import OutputFixingParser
from langchain_core.runnables import RunnablePassthrough# 創建基礎解析器
base_parser = PydanticOutputParser(pydantic_object=Product)# 添加錯誤處理
robust_parser = OutputFixingParser.from_llm(parser=base_parser,llm=ChatOpenAI()
)# 實現錯誤捕獲邏輯
def safe_parse(text):try:return robust_parser.parse(text)except Exception as e:return {"error": str(e), "raw_text": text}# 在鏈中使用安全解析
robust_chain = prompt | model | RunnablePassthrough.assign(parsed_output=lambda x: safe_parse(x.content)
)

與其他組件集成

1. 在智能體中使用解析器

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.tools import tool# 定義工具函數
@tool
def search_products(query: str) -> str:"""搜索產品數據庫"""# 假設實現return "找到3個產品..."# 定義輸出格式
class SearchResult(BaseModel):products: List[str] = Field(description="產品列表")top_pick: Optional[str] = Field(description="最佳推薦")reason: Optional[str] = Field(description="推薦理由")search_parser = PydanticOutputParser(pydantic_object=SearchResult)# 創建智能體
tools = [search_products]
agent = create_openai_tools_agent(model, tools, """你是一位產品專家，幫助用戶查找產品。最終輸出必須是JSON格式，包含產品列表和推薦。{format_instructions}""".format(format_instructions=search_parser.get_format_instructions())
)
agent_executor = AgentExecutor(agent=agent, tools=tools)# 處理用戶查詢
response = agent_executor.invoke({"input": "查找價格低于100元的手機配件"})
structured_result = search_parser.parse(response["output"])

2. 與檢索系統集成

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings# 假設已經有一個向量數據庫
vector_db = Chroma(embedding_function=OpenAIEmbeddings())
retriever = vector_db.as_retriever()# 定義輸出格式
class RetrievalSummary(BaseModel):question: str = Field(description="用戶問題")sources: List[str] = Field(description="信息來源")answer: str = Field(description="基于來源的回答")confidence: float = Field(description="置信度 (0.0-1.0)")# 創建解析器
rag_parser = PydanticOutputParser(pydantic_object=RetrievalSummary)# 格式化檢索文檔
def format_docs(docs):return "\n\n".join([f"來源 {i+1}: {doc.page_content}" for i, doc in enumerate(docs)])# 創建RAG鏈
rag_chain = {"question": lambda x: x["question"],"sources": lambda x: format_docs(retriever.get_relevant_documents(x["question"]))
} | ChatPromptTemplate.from_template("""基于以下來源回答問題。如果來源中沒有答案，就說你不知道。問題: {question}來源:{sources}{format_instructions}"""
) | model | rag_parser