LangGraph結構化輸出詳解:讓智能體返回格式化數據
引言
在使用大語言模型進行開發時,我們經常需要模型返回特定格式的數據,而不是純文本。比如在構建AI應用時,我們可能需要模型返回JSON格式的數據用于后續處理,或者返回符合特定數據結構的對象。這就是結構化輸出的價值所在。
本文將深入探討LangGraph中的結構化輸出功能,重點介紹.with_structured_output()方法的使用,并通過實際代碼示例展示如何在項目中應用這一技術。
什么是結構化輸出?
結構化輸出是指LLM(或LangGraph工作流中的其他節點)返回的輸出被格式化為明確定義的、機器可讀的對象,如字典、列表、自定義類或JSON對象,而不是純文本。
對比示例:
- 非結構化輸出:"答案是北京。"
- 結構化輸出:{"城市": "北京", "國家": "中國", "人口": 2154}
為什么需要結構化輸出?
在LangGraph中,一個節點的輸出通常作為另一個節點的輸入。如果輸出是結構化的,下一步就能準確知道期望什么字段/數據以及如何訪問它們。這大大提高了系統的可靠性和可維護性。
核心技術:.with_structured_output()方法
.with_structured_output()是獲得結構化輸出的最簡單、最可靠的方法。它專為那些原生支持結構化輸出API(如工具/函數調用或JSON模式)的模型實現,并在后臺使用這些功能。
基本工作原理
該方法接受一個模式作為輸入,指定所需輸出屬性的名稱、類型和描述。它返回一個類似于模型的Runnable,但輸出的不是字符串或消息,而是對應于給定模式的對象。
實現方式詳解
方式一:使用Pydantic類
使用Pydantic的主要優點是模型生成的輸出將被驗證。如果任何必需的字段缺失或任何字段的類型錯誤,Pydantic將拋出錯誤。
from pydantic import BaseModel, Field
from langchain_deepseek import ChatDeepSeek
from typing import Optional
# 定義Pydantic模型
class ProductInfo(BaseModel):
"""產品基本信息"""
name: str = Field(descriptinotallow="產品名稱")
price: float = Field(descriptinotallow="產品價格,單位為元")
category: str = Field(descriptinotallow="產品所屬類別")
rating: Optional[int] = Field(
default=None, descriptinotallow="產品評分,1-10分"
)
# 初始化模型
llm = ChatDeepSeek(model="deepseek-chat")
structured_llm = llm.with_structured_output(ProductInfo)
# 調用示例
result = structured_llm.invoke("幫我分析一下iPhone 15這款手機")
print(result)
# 輸出: ProductInfo(name='iPhone 15', price=5999.0, category='智能手機', rating=9)方式二:使用TypedDict
如果您不想使用Pydantic驗證,或者希望能夠流式輸出模型結果,可以使用TypedDict類定義模式。
from typing import Optional
from typing_extensions import Annotated, TypedDict
# 使用TypedDict定義模式
class StudentInfo(TypedDict):
"""學生基本信息"""
name: Annotated[str, ..., "學生姓名"]
age: Annotated[int, ..., "學生年齡"]
major: Annotated[str, ..., "所學專業"]
score: Annotated[Optional[float], None, "平均成績,0-100分"]
structured_llm = llm.with_structured_output(StudentInfo)
result = structured_llm.invoke("張明,20歲,計算機科學專業,平均成績85分")
print(result)
# 輸出: {'name': '張明', 'age': 20, 'major': '計算機科學', 'score': 85.0}方式三:使用JSON Schema
您也可以直接傳入JSON Schema字典,這種方式不需要導入額外的類,但代碼會更冗長一些。
json_schema = {
"title": "公司信息",
"description": "公司基本信息",
"type": "object",
"properties": {
"company_name": {
"type": "string",
"description": "公司名稱",
},
"industry": {
"type": "string",
"description": "所屬行業",
},
"employee_count": {
"type": "integer",
"description": "員工總數",
"default": None,
},
},
"required": ["company_name", "industry"],
}
structured_llm = llm.with_structured_output(json_schema)
result = structured_llm.invoke("分析一下騰訊公司")
print(result)
# 輸出: {'company_name': '騰訊', 'industry': '互聯網科技', 'employee_count': 110715}LangGraph中的完整應用示例
下面是一個完整的LangGraph應用示例,展示如何在工作流中使用結構化輸出:
import json
from pydantic import BaseModel, Field
from langchain_core.messages import HumanMessage, AIMessage
from langchain_deepseek import ChatDeepSeek
from langgraph.graph import StateGraph, MessagesState, END
from dotenv import load_dotenv
load_dotenv()
# 定義結構化輸出模型
class AnalysisResult(BaseModel):
"""文本分析結果"""
topic: str = Field(descriptinotallow="文本主要主題")
summary: str = Field(descriptinotallow="內容摘要,不超過100字")
keywords: list[str] = Field(descriptinotallow="提取的關鍵詞列表")
sentiment: str = Field(descriptinotallow="情感傾向:積極/消極/中性")
# 初始化模型
model = ChatDeepSeek(model="deepseek-chat")
structured_llm = model.with_structured_output(AnalysisResult).with_config(tags=["文本分析器"])
# 定義分析節點
def analysis_node(state: MessagesState):
"""對輸入文本進行結構化分析"""
result = structured_llm.invoke(state["messages"])
print(f"分析結果: {result}")
# 將分析結果轉換為AIMessage
response_content = f"""
?? 分析報告
主題: {result.topic}
摘要: {result.summary}
關鍵詞: {', '.join(result.keywords)}
情感傾向: {result.sentiment}
""".strip()
return {"messages": [AIMessage(cnotallow=response_content)]}
# 構建LangGraph工作流
workflow = StateGraph(MessagesState)
workflow.add_node("分析", analysis_node)
workflow.set_entry_point("分析")
workflow.add_edge("分析", END)
# 編譯圖
graph = workflow.compile()
# 測試用例
def test_analysis():
"""測試文本分析功能"""
test_text = """
人工智能技術正在快速發展,特別是大語言模型的出現,
為各個行業帶來了前所未有的變革機會。從自動化客服到智能寫作,
從代碼生成到數據分析,AI正在重塑我們的工作方式。
雖然技術發展令人興奮,但我們也需要關注AI倫理和安全性問題。
"""
graph_input = {"messages": [HumanMessage(cnotallow=f"請分析以下文本:{test_text}")]}
# 運行分析
result = graph.invoke(graph_input)
print("="*50)
print("最終輸出:")
print(result["messages"][-1].content)
# 執行測試
if __name__ == "__main__":
test_analysis()多模式選擇
有時我們需要讓模型在多個輸出格式之間進行選擇。最簡單的方法是創建一個具有Union類型屬性的父模式:
from typing import Union
from pydantic import BaseModel, Field
class Joke(BaseModel):
"""笑話內容"""
setup: str = Field(descriptinotallow="笑話的鋪墊部分")
punchline: str = Field(descriptinotallow="笑話的笑點部分")
rating: Optional[int] = Field(default=None, descriptinotallow="有趣程度,1-10分")
class ConversationalResponse(BaseModel):
"""普通對話回應"""
response: str = Field(descriptinotallow="對用戶查詢的對話式回應")
class FinalResponse(BaseModel):
"""最終輸出,可以是笑話或普通回應"""
final_output: Union[Joke, ConversationalResponse]
structured_llm = llm.with_structured_output(FinalResponse)
# 測試不同類型的輸入
print("=" * 30)
print("請求笑話:")
result1 = structured_llm.invoke("給我講個關于程序員的笑話")
print(result1)
print("=" * 30)
print("普通對話:")
result2 = structured_llm.invoke("你今天怎么樣?")
print(result2)流式輸出支持
當輸出類型是字典時(使用TypedDict類或JSON Schema定義時),我們可以從結構化模型中進行流式輸出:
from typing_extensions import Annotated, TypedDict
class NewsSummary(TypedDict):
"""新聞摘要信息"""
title: Annotated[str, ..., "新聞標題"]
summary: Annotated[str, ..., "新聞摘要內容"]
keywords: Annotated[list[str], ..., "新聞關鍵詞"]
structured_llm = llm.with_structured_output(NewsSummary)
print("流式輸出示例:")
for chunk in structured_llm.stream("總結今天的科技新聞"):
print(chunk)高級技巧
1. 原始輸出處理
通過設置include_raw=True,可以獲取原始輸出,便于調試和錯誤處理:
structured_llm = llm.with_structured_output(AnalysisResult, include_raw=True)
result = structured_llm.invoke("分析人工智能發展")
print("原始輸出:", result['raw'])
print("解析結果:", result['parsed'])
print("解析錯誤:", result['parsing_error'])2. 指定輸出方法
對于支持多種結構化輸出方法的模型,可以通過method參數指定使用的方法:
# 使用JSON模式
structured_llm = llm.with_structured_output(AnalysisResult, method="json_mode")最佳實踐
- 選擇合適的模式類型:
- 需要運行時驗證 → 使用Pydantic的BaseModel
- 僅需靜態類型檢查 → 使用TypedDict
- 需要流式輸出 → 使用TypedDict或JSON Schema
- 提供清晰的字段描述:字段的名稱和描述對模型理解輸出格式非常重要
- 使用中文字段名:在中文應用場景中,使用中文字段名可以提高模型的理解準確性
- 合理設置可選字段:使用Optional類型和默認值來處理不確定的信息
總結
結構化輸出是現代AI應用開發中的重要技術。通過LangGraph的.with_structured_output()方法,我們可以輕松讓大語言模型返回格式化的數據,提高應用的可靠性和可維護性。
無論是使用Pydantic進行嚴格的數據驗證,還是使用TypedDict進行靈活的字典操作,或是直接使用JSON Schema,都能滿足不同場景下的需求。關鍵是根據具體的應用場景選擇最合適的實現方式。
在實際項目中,建議優先使用Pydantic方式,因為它提供了最好的類型安全和數據驗證功能。對于需要流式輸出的場景,則可以考慮使用TypedDict方式。
通過合理運用這些技術,我們可以構建更加健壯和高效的AI應用系統。























