iT邦幫忙

2026 iThome 鐵人賽

DAY 3
0
Vibe Coding

agent工作流系列 第 7

【Day 7】怎麼應用 Structured Output Engineering:Pydantic 契約、錯誤修復與資料庫無縫寫入

  • 分享至 

  • xImage
  •  

在昨天的文章中,我們探討了Structured Output Engineering的核心本質——透過約束解碼與Schema約束,將LLM原生發散的文字輸出收斂成100% 型別安全(Type-safe)的結構化資料

今天我們進入實戰篇:在真實專案中,我們該如何使用 Pydantic 定義資料契約、透過LangChain / OpenAI SDK啟用嚴格模式(Strict Mode),並將模型輸出的資料無縫寫入下游資料庫?


一、實戰場景:電商客服對話抽取與資料庫入庫

假設我們的工作流需要從非結構化的客戶對話中,自動提取以下資訊並寫入資料庫:

  1. 客戶意圖(Intent):僅限退貨、換貨或物流查詢
  2. 訂單編號(Order ID):若有提及則提取,否則為空
  3. 退換貨原因標籤(Reason Tags):字串陣列
  4. 急迫程度評分(Urgency Score):整數1到5

二、核心實作 1:使用 Pydantic 定義資料契約(Schema Contract)

在Python中,Pydantic是定義結構化輸出的黃金標準
透過型別提示與 Field 描述,我們不僅建立了資料驗證規則,同時也完成了對模型的提示詞引導

from typing import List, Optional, Literal
from pydantic import BaseModel, Field

class OrderActionSchema(BaseModel):
    intent: Literal["RETURN", "EXCHANGE", "TRACKING", "OTHER"] = Field(
        description="客戶的核心業務意圖"
    )
    order_id: Optional[str] = Field(
        default=None,
        description="提取出的訂單編號(例如 ORD-12345),若未提及則為 None"
    )
    reason_tags: List[str] = Field(
        default_factory=list,
        description="提取出的具體原因關鍵詞標籤清單,例如 ['尺寸不合', '外包裝破損']"
    )
    urgency_score: int = Field(
        ge=1, le=5,
        description="客戶情緒的急迫程度評級,範圍為 1(不急)到 5(極度緊急)"
    )

三、核心實作 2:利用 LangChain / OpenAI 啟用嚴格結構化輸出

現代SDK多已內建 .with_structured_output() 方法,能底層啟用 json_schema 嚴格約束(Strict Mode),直接回傳已反序列化完成的 Pydantic 物件

import os
from langchain_openai import ChatOpenAI

# 初始化支援結構化輸出的模型
llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0.0
)

# 綁定 Pydantic Schema,開啟嚴格結構化解析
structured_extractor = llm.with_structured_output(
    OrderActionSchema,
    method="json_schema",
    strict=True
)

# 模擬客服對話文字
user_chat = """
您好,我三天前買的衣服送來了,訂單編號是 ORD-8899。
但是寄來的顏色完全寄錯,而且明天我出國就要穿,麻煩你們務必今天幫我處理換貨!真的非常急!
"""

# 執行結構化抽取
result: OrderActionSchema = structured_extractor.invoke(
    f"請從以下對話中抽取結構化資訊:\n\n{user_chat}"
)

# 輸出結果保證為型別安全的 Pydantic 物件
print(type(result))            # <class '__main__.OrderActionSchema'>
print(result.intent)          # EXCHANGE
print(result.order_id)        # ORD-8899
print(result.reason_tags)     # ['顏色寄錯', '急需出國使用']
print(result.urgency_score)   # 5

四、核心實作 3:下游無縫整合——寫入 SQLite 資料庫

因為輸出已經是標準的Python物件,下游程式碼可以直接進行型別檢查、條件分支判斷,並直接與ORM或資料庫驅動整合,完全不需要撰寫脆弱的Regex或手動 json.loads()

import sqlite3

def init_db():
    conn = sqlite3.connect("support_system.db")
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS support_tickets (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            order_id TEXT,
            intent TEXT,
            reasons TEXT,
            urgency INTEGER
        )
    """)
    conn.commit()
    conn.close()

def save_extracted_ticket(ticket: OrderActionSchema):
    conn = sqlite3.connect("support_system.db")
    cursor = conn.cursor()
    
    # 直接使用型別安全的屬性寫入資料庫
    cursor.execute("""
        INSERT INTO support_tickets (order_id, intent, reasons, urgency)
        VALUES (?, ?, ?, ?)
    """, (
        ticket.order_id,
        ticket.intent,
        ", ".join(ticket.reason_tags),
        ticket.urgency_score
    ))
    
    conn.commit()
    conn.close()
    print(f"成功入庫:訂單 {ticket.order_id},意圖 {ticket.intent},緊急度 {ticket.urgency_score}")

# 初始化並儲存
init_db()
save_extracted_ticket(result)

五、實戰進階:解析防禦與自動修復管線(Validation & Retry Pipeline)

在某些不支援原生 json_schema 嚴格約束的小型開源模型上,若偶發輸出格式不符,我們可以透過Pydantic的 ValidationError 結合重試管線(Retry Parser)進行自我修復:

[LLM 輸出非標準 JSON / 欄位遺失]
             │
             ▼
   [Pydantic 驗證失敗] ───> 攔截 ValidationError
             │
             ▼
[將原始錯誤訊息回傳給 LLM]
"你輸出的 JSON 缺少了 'urgency_score' 欄位,請修正並重新輸出。"
             │
             ▼
     [LLM 自動修復並重新輸出]

六、實戰總結:結構化輸出工程架構檢視

環節 傳統做法 (Regex / 手動 JSON) 結構化輸出做法 (Pydantic / Strict Mode)
定義方式 Prompt 自然語言描述 程式碼級別 Pydantic Model / JSON Schema
型別安全 執行期充滿 KeyError 風險 編譯期與執行期皆有型別提示與驗證
資料流銜接 需手動寫 Parser 解析與清理 直接調用 .intent.dict() 對接 ORM
邊界限制 模型可能輸出未定義的分類字串 LiteralEnum 物理級別鎖死枚舉範圍

小結與下集預告

掌握了Structured Output Engineering,我們徹底打通了從「LLM 文字生成」到「傳統軟體與資料庫」之間的型別壁壘,讓 LLM 能作為可靠的資料轉換模組存在於系統中

然而,單純讓模型「輸出結構化資料」還不足以讓它成為自主運作的Agent
真正的Agent還必須能夠主動**「調用外部系統、查詢即時天氣、打 API 獲取即時資料」**。

明天 【Day 8】甚麼是 Tool Calling Engineering,我們將探討現代 Agent 如何透過 Function Calling / Tool Calling 具備「使用工具、干涉外部世界」的能力!


上一篇
【Day 6】甚麼是 Structured Output Engineering:讓模型輸出從「純文字」邁向「型別安全」
下一篇
【Day 8】甚麼是 Tool Calling Engineering:賦予LLM連結真實世界的手與腳
系列文
agent工作流8
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言