Schema 驅動的決策
把一份 JSON schema 或一個 pydantic 模型變成 Laya 問題,拿回帶校準置信度的型別化值。正是這座橋 讓 Laya 成為一個結構化輸出引擎:你描述想要的形狀,Laya 在一次前向傳播裡給出答案。
import laya
schema = {
"type": "object",
"properties": {
"department": {"type": "string", "enum": ["billing", "support", "sales"],
"description": "Which team should handle this?"},
"urgency": {"type": "integer", "minimum": 0, "maximum": 2},
"needs_human": {"type": "boolean"},
},
}
agent = laya.load("convaiinnovations/laya")
values = agent.decide("I was charged twice, refund me.", schema=schema)
# {"department": "billing", "urgency": 2, "needs_human": True}
用 pydantic(安裝 laya[structured]):
from typing import Literal
from pydantic import BaseModel
class Ticket(BaseModel):
department: Literal["billing", "support", "sales"]
urgency: Literal[0, 1, 2]
needs_human: bool
ticket = agent.decide("I was charged twice, refund me.", schema=Ticket)
支援的子集
頂層必須是一個帶 properties 的物件。每個屬性變成一個 Laya 問題。
下面每一行都是真實的 schema:tests/test_structured_docs.py 編譯第一列,並斷言編譯器實際產出的
問題,所以這張表不會和程式碼脫節。一個單元格要麼是單獨的屬性 schema,要麼是對某個入口點的呼叫。
| JSON schema | Laya 問題 | 返回的值 |
|---|---|---|
{"enum": ["billing", "support"]} |
choice |
被選中的值,保留它原本的型別 |
{"const": "billing"} |
choice |
那一個值 |
{"type": "boolean"} |
noul |
true / false |
{"type": "integer", "minimum": 0, "maximum": 5} |
score |
機率最高的檔位,作為一個整數 |
{"type": "number", "minimum": 0, "maximum": 5} |
score |
那個檔位,作為一個整數 |
{"type": "string", "enum": ["low", "high"], "description": "How urgent?"} |
choice |
How urgent? 就是問題措辭 |
{"anyOf": [{"enum": ["x", "y"]}, {"type": "null"}]} |
choice |
和普通的 enum 行一樣;沒有答案時這個鍵不出現 |
{"oneOf": [{"type": "boolean"}, {"type": "null"}]} |
noul |
和普通的 boolean 行一樣 |
{"type": ["integer", "null"], "minimum": 1, "maximum": 3} |
score |
和普通的帶界整數行一樣 |
Literal[...] 和 Optional[...] 是 pydantic 裡對應 enum 和 anyOf 兩行的寫法:
questions_from_pydantic 把它們渲染成那些形狀,同樣的行適用。
title 不會被讀取。pydantic 不管你有沒有要求,都會給 model_json_schema() 的每個欄位放上
一個,而一個按屬性的名字沒法給一個問題所據以構建的那些逐選項 choice 打標籤,所以措辭的槓桿是
description —— 見下面的它內部如何對映。
投影是精確的:enum: [1, 2, 3] 返回 2,不是 "2";帶界整數返回 minimum 和 maximum
之間的一個檔位;布林值就是 noul >= 0.5。
拒絕
一個無法從固定選項集作答的 schema 會丟擲 laya.structured.SchemaError(一個 ValueError),
並點出確切的路徑。每一行也都會被執行,欄位名取 name:
| 屬性 schema | 訊息 |
|---|---|
{"type": "string"} |
properties.name: a free string cannot be a fixed option set; use 'enum' or a boolean |
{"type": "array", "items": {"type": "string"}} |
properties.name: arrays are not supported; ask one field per element |
{"type": "object", "properties": {"inner": {"type": "boolean"}}} |
properties.name: nested objects are not supported; flatten the schema |
{"$ref": "#/definitions/node"} |
properties.name: $ref/recursion is not supported; flatten the schema |
{"enum": [1, "1"]} |
properties.name: enum values produce duplicate choice labels |
{"enum": []} |
properties.name: 'enum' must not be empty |
{"type": "number"} |
properties.name: a numeric field needs integer 'minimum' and 'maximum' to become a score |
{"type": "integer", "minimum": 5, "maximum": 2} |
properties.name: 'maximum' 2 is below 'minimum' 5 |
{"type": "integer", "minimum": 0, "maximum": 10} |
properties.name: 11 levels exceeds MAX_SCORE_LEVELS=10; narrow the range or use an enum |
{"anyOf": [{"type": "string"}, {"type": "integer"}]} |
properties.name: only 'Optional[...]' unions (one non-null branch) are supported, got 2 |
{"type": ["string", "integer"]} |
properties.name: 'type' has multiple non-null types; unions are not supported |
{"format": "date"} |
properties.name: unsupported schema {'format': 'date'} |
"boolean" |
properties.name: property must be an object, got str |
入口點自身會拒絕這些:
| 呼叫 | 訊息 |
|---|---|
plan_from_json_schema("not a schema") |
expected a JSON schema object, got str |
plan_from_json_schema({"type": "object"}) |
the top level must be an object with 'properties' |
plan_from_json_schema({"type": "object", "properties": {}}) |
'properties' must be a non-empty object |
plan_from_json_schema({"type": "object", "properties": {"p%d" % i: {"type": "boolean"} for i in range(33)}}) |
33 properties exceeds MAX_PROPERTIES=32 |
plan_from_json_schema({"type": "object", "properties": {"name": {"enum": ["v%d" % i for i in range(33)]}}}) |
properties.name: 33 options exceeds MAX_OPTIONS=32 |
decide(None, "I was charged twice.", schema=42) |
expected a JSON schema dict or a pydantic model, got int |
限制:MAX_PROPERTIES = 32、MAX_OPTIONS = 32、MAX_SCORE_LEVELS = 10。
API
| 函式 | 用途 |
|---|---|
laya.decide(runner, state, schema=..., *, questions=..., return_details=..., min_confidence=..., **predict_kwargs) |
自由函式,對 Agent 和 Router 都適用 |
agent.decide(state, schema=..., ...) / router.decide(state, schema=..., ...) |
便捷方法 |
laya.decide_batch(runner, states, schema=..., ...) / agent.decide_batch(...) / router.decide_batch(...) |
在多個狀態上做同樣的事,一次批呼叫 |
questions_from_json_schema(schema) |
schema 轉成 Laya 問題 |
questions_from_pydantic(model) |
pydantic 模型轉成問題(需要 pydantic) |
answers_to_json(answers, schema) |
把原始答案投影到 schema 的值上 |
answer_to_pydantic(model, answers) |
把原始答案投影成一個 pydantic 例項 |
plan_from_json_schema(schema) |
經過校驗的欄位計劃(進階) |
schema 和 questions 只能傳其中一個。傳 questions 時,decide 返回原始答案而不是做投影。
額外的關鍵字參數會轉發給 predict,所以鉤子、model=、task= 和 token 預算都能用:
router.decide(state, schema=Ticket, model="multilingual", hooks=[Metrics()])
給多個狀態打分
decide_batch 是吞吐的形式:schema 只規劃一次,它的問題通過 predict_batch 在每一個狀態上跑,
於是多個狀態共享前向傳播,而不是每次呼叫一次。結果按輸入順序返回,投影方式和 decide 完全一樣,
而 return_details=True 會給每個狀態一個 DecisionResult:
values = agent.decide_batch(ticket_texts, schema=Ticket) # values[i] matches ticket_texts[i]
results = router.decide_batch(states, schema=Ticket, return_details=True, batch_size=64)
在 Router 上,每個狀態仍然各自路由,所以一次呼叫可以橫跨多個 checkpoint。關鍵字參數會到達
predict_batch,所以 batch_size=、model= 和鉤子都和 decide 一樣能用。Agent、ONNXAgent
和 Router 都有它;沒有 predict_batch 的 runner 會拋 TypeError,而不是靜默回退成一個迴圈
—— 那種情況就逐狀態呼叫 decide。批處理會像 predict_batch 那樣讓臨界 argmax 發生偏移;README
記錄了兩種裝置上實測的加速。
置信度和機率
預設 decide 只回傳值。傳 return_details=True 得到一個 DecisionResult,帶逐欄位的置信度、
機率、原始答案,以及這次呼叫的 usage 和路由:
result = agent.decide(state, schema=Ticket, return_details=True)
result.values["department"] # "billing"
result.answer_confidence["department"] # 0.94 max(p): the quantity min_confidence gates on
result.confidence["department"] # 0.71 normalized entropy, which depends on label count
result.probabilities["department"] # {"billing": 0.94, "support": 0.06, "sales": 0.0}
result.usage # {"input_tokens": ..., "output_tokens": 0,
# "state_tokens": ..., "state_tokens_dropped": ...,
# "truncated": ..., "truncated_questions": [...]}
result.routing # the Router decision, when a Router answered
usage 是 predict() 構建出來的那個塊,原樣轉發:input_tokens 把一個 state 上每個問題各一行的量加總,而 output_tokens 永遠是 0,因為什麼都不生成;其餘四個鍵是截斷報告(#174)—— state_tokens 是整個序列化 state 所需的量,state_tokens_dropped 則是任何單個問題的 head 所放棄的最大量,truncated / truncated_questions 則說明是哪一個。第七個鍵 options,只在 head 預算讓某個問題的選項共享了一段 token 範圍(#538)時才出現;docs/http-api.md 把這個塊和那個欄位都記錄為響應鍵,而 tests/test_structured_docs.py 把這一頁的清單釘在構建它的程式碼上。
confidence 和 answer_confidence 是兩個不同的量,名字本身就說明了各自的定義。
answer_confidence 是 max(p),即報告出來的那個答案所分到的機率質量。它正是溫度縮放所擬合的
東西,是本倉庫裡每一個校準數字所據以計算的東西,也是 min_confidence 所比較的物件 —— 這正是
該對它、而不是對 confidence 做門控的原因。confidence 是歸一化熵,它取決於問題當時有多少個
選項:tests/test_confidence.py 固定了這一點 —— 一個兩選項分佈在一個 noul 上返回 0.90,在
一個等價的 choice 上返回 0.53 —— 所以它不能拿去和閾值比較。一個沒有報告出可用
answer_confidence 的欄位會對映到 None,這和報告出一個 0.0 不是一回事。
門控要和門控本身所用的量一致:
if result.answer_confidence["department"] < 0.6:
result.values["department"] = "human-review"
answer_confidence 適合用來篩選,這不等於它就是一個可信的機率。把它讀作「在 c 處返回的答案裡
約有 c 的比例是對的」,只有在溫度已經擬合、並且針對該 checkpoint 和問題形狀在留出資料上驗證過
之後才成立。隨包釋出的 checkpoint 按釋出狀態看是過度自信的,而 laya-multilingual 乾脆沒有自帶
任何已擬合的溫度 —— 見 README 的
校準 和
誠實的限制 兩節,以及
微調 notebook
裡的擬合迴圈。在依賴這個級別之前先擬合;報告它,是因為門控和評估 harness 用的都是它。
它內部如何對映
- Enum 和
Literal變成choice問題,值作為字串標籤;標籤在返回時映射回原來的值,所以整數 仍然是整數。 - 帶界整數變成一個
score問題,每個值一個檔位;返回的值是minimum + argmax。 - 布林值變成一個
noul問題;值為noul >= 0.5。 description變成問題的指令,所以一個好的描述正是決策準確的原因。這遵循與鉤子指南 相同的規則:把每個選項的含義說清楚。null分支在欄位被規劃之前就被丟掉,所以Optional[X]問的正是X問的問題。沒有答案時,該 欄位的鍵就直接從值裡缺席,這正是能安全地把一個欄位宣告為可選、又不改變模型所見內容的原因。
另見
- 預測鉤子:觀察、塑造、快取或門控它產生的這些決策。
- 決策原語:深入講
choice、score和noul。 - LangChain 與 LangGraph:
LayaDecision就是這次呼叫作為一個 runnable 放進鏈裡。