第 7 章:应用构建:从原语到完整系统
第 7 章:应用构建:从原语到完整系统
本文整理自 Datawhale 开源项目 datawhalechina/jev-cookbook(CC BY-NC-SA 4.0),源文件:main/02_核心概念/05_应用构建.ipynb。
针对官方文档对应章节的可运行实验笔记,全部实验使用中文场景与中文提示词。 面向会基础 Python、刚接触 AI Agent 的读者。
学习目标: 复刻官方客服分流,拆解凭据信号,记录每个分支的覆盖情况,并练习结构化问题。
官方原文 · 中文参考。本章以中文重述理论、复刻对应场景;扩展实验会单独说明。 所有客户、订单及消息均为教学合成数据。
笔记本结构
| 章节 | 内容 |
|---|---|
| 0. 准备 | 安装库、配置客户端、连通性测试与离线示例 |
| 1 | 架构与确定性规则 |
| 2 | 中文客服工单的七个问题 |
| 3 | 多场景试句与分支覆盖 |
| 4 | 航班、嵌套状态、候选记录、虚拟卡与工具轨迹 |
| 练习与小结 | 练习、自查、总结与本次执行记录 |
实验按原理 → 理论根基 → 定义数据 → 定义问题 → 调用 → 解读结果展开,每个代码单元格只做一件事。
运行要求
- Python ≥ 3.10;本章使用
typesafe-sdk==0.7.0。 - 真实实验需要启动进程的
TYPESAFE_API_KEY环境变量,密钥不要写进 Notebook。
在本仓库 notebooks/ 目录创建环境并打开本文件:
./setup_env.sh
.venv/bin/python -m pip install -r requirements.txt -c generators/constraints-foundations.txt
.venv/bin/jupyter lab 08_应用构建.ipynb产品名、字段名和选项 key 保持英文,state、提示词与解说使用中文。
默认 JEV_RUN_MODE=auto(在线优先):检测到 TYPESAFE_API_KEY 即调用真实模型;未检测到才回退离线替身,每格输出都标注来源。
正式验收设置 JEV_RUN_MODE=live(无密钥直接报错、不回退);强制纯离线学习设置 JEV_RUN_MODE=offline。
验证状态:2026-09-25 已用真实 API 在线执行本章全部调用,下方输出即实测结果。
批量执行、离线预览和验收记录见本目录 MAINTENANCE.md。
0. 准备
本节可折叠阅读,但独立运行时不能跳过。客户端、辅助对象和示例数据都在本文件中定义。
0.1 安装依赖
推荐先运行 setup_env.sh。只有当前内核缺少 SDK 时,本格才安装依赖。
import importlib.util
if importlib.util.find_spec("typesafe_sdk") is None:
%pip install -q typesafe-sdk==0.7.0观察与理解: 安装包的名字是 typesafe-sdk,Python 导入名是 typesafe_sdk。安装成功不代表 API 已连通。
0.2 导入与配置
默认模型固定版本,便于记录实验条件;可通过环境变量更换。不要从 Notebook 输入密钥。
import os
import json
import time
import math
from datetime import datetime, timezone
from importlib.metadata import version
from typesafe_sdk import (
Choice, Score, Noul, NoulCriteria, TypeSafeClient,
TypeSafeAuthenticationError, RetryPolicy,
)
MODEL = os.environ.get("TYPESAFE_DEFAULT_MODEL", "jev-1.13.0")
RUN_MODE = os.environ.get("JEV_RUN_MODE", "auto")
API_KEY = os.environ.get("TYPESAFE_API_KEY", "")
if RUN_MODE not in {"live", "offline", "auto"}:
raise ValueError("JEV_RUN_MODE 只能是 live、offline 或 auto")
if RUN_MODE == "live" and not API_KEY:
raise RuntimeError("JEV_RUN_MODE=live 需要真实密钥:请配置 TYPESAFE_API_KEY;仅学习可改用默认 auto(无密钥自动离线)")
client = None if (RUN_MODE == "offline" or not API_KEY) else TypeSafeClient(
api_key=API_KEY, model=MODEL, timeout=30, retry=RetryPolicy(max_retries=0))
mode_note = ("在线优先:本次会话调用真实模型" if client is not None else
("强制离线(JEV_RUN_MODE=offline)" if RUN_MODE == "offline" else
"未检测到 TYPESAFE_API_KEY → 离线替身;配置密钥后重跑本格即切在线实测"))
print("模式:", RUN_MODE, "|", mode_note, "|SDK:", version("typesafe-sdk"), "|模型配置:", MODEL)模式: auto | 在线优先:本次会话调用真实模型 |SDK: 0.7.1 |模型配置: jev-1.13.0默认在线优先:有密钥即走真实模型,每次调用的来源(live/offline)都记录在 CALL_LOG 与输出中;auto 仅在未配置密钥或 401 时回退离线替身。正式验收设 JEV_RUN_MODE=live(禁用回退、不自动重试,请求数量有界)。
0.3 连通性测试
用一条 Noul 检查真实响应能否返回。网络、限流与输入错误直接抛出,不伪装成不确定判断。
PING = {"source": "offline", "reason": "未发起连通性请求(无 client:未配置密钥或强制离线)"}
if client is not None:
try:
ping = client.system_one("你好", {"greeting": Noul(
instructions="这段文字是否在打招呼?")})
PING = {"source": "live", "model": ping.model,
"input_tokens": ping.usage.input_tokens,
"output_tokens": ping.usage.output_tokens}
except TypeSafeAuthenticationError:
if RUN_MODE == "live":
raise
client.close()
client = None
PING["reason"] = "401 鉴权失败,仅教学模式允许回退"
print(json.dumps(PING, ensure_ascii=False)){"source": "live", "model": "jev-1.13.0", "input_tokens": 278, "output_tokens": 22}观察与理解: source=live 表示这一次连通性请求成功;仍要查看后续实验记录,不能用它代替整章验收。
0.4 离线替身
沿用参考模板的 _FakeAnswer 与 _FakeResponse 访问方式。人工数字仅用来检验读取字段和代码分支。
class _FakeAnswer:
def __init__(self, type_, **values):
self.type = type_
for name, value in values.items():
setattr(self, name, value)
class _FakeResponse:
def __init__(self, answers):
self.answers = answers
self.choices = {k: v for k, v in answers.items() if v.type == "choice"}
self.scores = {k: v for k, v in answers.items() if v.type == "score"}
self.nouls = {k: v for k, v in answers.items() if v.type == "noul"}
self.model = "人工示例,非模型预测"
self.usage = _FakeAnswer("usage", input_tokens=0, output_tokens=0)人工 Score 由概率计算期望,避免模板中的分数与分布不一致。人工 confidence 只是指定的演示字段,不是在复现服务端的计算公式。
定义两种示例答案构造器;Noul 可直接用 _FakeAnswer。所有具体答案集中在下一节。
def fake_choice(probabilities, confidence):
return _FakeAnswer("choice", choice=max(probabilities, key=probabilities.get),
probabilities=probabilities, confidence=confidence)
def fake_score(probabilities, legend, confidence):
return _FakeAnswer("score", score=sum(k * p for k, p in probabilities.items()),
probabilities=probabilities, legend=dict(enumerate(legend)),
confidence=confidence)观察与理解: 例如概率 {0:0.05, 1:0.26, 2:0.69} 对应 1.64。不能把另一个数与该分布配在一起。
0.5 统一调用入口
每次调用记录来源、模型与 token 用量。离线耗时记为 None,不把本地字典访问当成模型速度。
CALL_LOG = []
class TS:
def call(self, state, questions, offline_answers, label):
start = time.perf_counter()
source = "live"
if client is None:
response, source = _FakeResponse(offline_answers), "offline"
else:
try:
response = client.system_one(state, questions)
except TypeSafeAuthenticationError:
if RUN_MODE != "auto":
raise
response, source = _FakeResponse(offline_answers), "offline"
validate_response(response, questions)
CALL_LOG.append({"case": label, "source": source, "model": response.model,
"seconds": time.perf_counter() - start if source == "live" else None,
"input_tokens": response.usage.input_tokens,
"output_tokens": response.usage.output_tokens})
if source == "offline":
print("离线替身(未调用真实模型):", label, ";人工答案,仅演示代码路径")
return response
ts = TS()与参考模板相比,这里增加了严格 live 模式和逐次记录。保留 401 教学回退,但超时、429 等错误继续失败,防止验收被回退掩盖。
校验结构与数值契约;只断言接口应满足的性质,不断言真实模型必须预测某个标签。
def validate_response(response, questions):
if set(response.answers) != set(questions):
raise ValueError("答案 ID 与问题 ID 不一致")
for key, question in questions.items():
answer = response.answers[key]
if isinstance(question, Noul):
if not 0 <= answer.noul <= 1:
raise ValueError("Noul 超出概率范围")
continue
probabilities = answer.probabilities
if not all(math.isfinite(p) and 0 <= p <= 1 for p in probabilities.values()):
raise ValueError("概率值无效")
if not math.isclose(sum(probabilities.values()), 1, abs_tol=0.02):
raise ValueError("概率之和偏离 1")
if not 0 <= answer.confidence <= 1:
raise ValueError("confidence 超出范围")
if isinstance(question, Choice):
if set(probabilities) != set(question.criteria):
raise ValueError("Choice 选项集合不一致")
if answer.choice not in probabilities:
raise ValueError("Choice 标签不在选项中")
else:
expected = sum(int(k) * p for k, p in probabilities.items())
if not math.isclose(answer.score, expected, abs_tol=0.03):
raise ValueError("Score 与概率加权期望不一致")观察与理解: 容差用于服务端数值舍入。结构检查通过只说明响应可读取,不证明语义判断正确。
显示结果时统一列出类型、概率和置信度;Noul 不额外制造 confidence 字段。
def show(response):
rows = {}
for key, answer in response.answers.items():
rows[key] = {name: getattr(answer, name) for name in
("type", "choice", "score", "noul", "confidence", "probabilities", "legend")
if hasattr(answer, name)}
print(json.dumps(rows, ensure_ascii=False, indent=2))0.6 本章离线示例数据
以下数值全部人工构造,专门测试分支;不来自 Jev,也不能用于估计中文准确率或校准情况。正式 live 运行不会使用这些答案。
def workflow_fixture(topic, confidence=0.93, risk=0.02, frustration=0):
other_p = (1 - confidence) / 2
return {
"topic": fake_choice({k: confidence if k == topic else other_p
for k in ["billing", "orders", "account"]}, confidence),
"requests_credentials": _FakeAnswer("noul", noul=risk),
"sender_identity_mismatch": _FakeAnswer("noul", noul=risk),
"unexpected_reward": _FakeAnswer("noul", noul=risk),
"refund_requested": _FakeAnswer("noul", noul=0.94 if topic == "billing" else 0.03),
"mentions_open_order": _FakeAnswer("noul", noul=0.95 if topic == "orders" else 0.04),
"frustration": fake_score({i: float(i == frustration) for i in range(3)},
["平静且客观", "不满但保持礼貌", "非常愤怒或威胁离开"], 0.93),
}
WORKFLOW_OFFLINE = {
"closed": workflow_fixture("billing"),
"billing": workflow_fixture("billing"),
"orders": workflow_fixture("orders"),
"account_normal": workflow_fixture("account"),
"account_high": workflow_fixture("account", frustration=2),
"quarantine": workflow_fixture("account", risk=0.96),
"review_spam": workflow_fixture("account", risk=0.5),
"review_topic": workflow_fixture("billing", confidence=0.42),
}这些人工答案特意覆盖代码的八条路径。真实调用不保证命中相同路径;本章末尾会列出实际观察到和未观察到的分支。
FLIGHT_OFFLINE = {"policy_supports_refund": _FakeAnswer("noul", noul=0.97)}
NESTED_OFFLINE = {
"duplicate_charge": _FakeAnswer("noul", noul=0.95),
"password_reset_supported": _FakeAnswer("noul", noul=0.96),
}
RECORD_OFFLINE = {"same_as_record_18": _FakeAnswer("noul", noul=0.94)}
CARD_OFFLINE = {"card_help_topic": fake_choice({
"get_disposable_virtual_card": 0.03, "disposable_card_limits": 0.97}, 0.95)}
TRACE_OFFLINE = {key: _FakeAnswer("noul", noul=probability) for key, probability in {
"geocode_tool_is_relevant": 0.98, "geocode_location_matches": 0.98,
"geocode_arguments_match_schema": 0.98, "geocode_result_matches_call": 0.98,
"weather_tool_is_relevant": 0.98, "weather_arguments_match_schema": 0.97,
"weather_uses_geocoded_coordinates": 0.98, "weather_date_matches": 0.98,
"weather_unit_matches": 0.02,
}.items()}1. 📖 理论根基:三种架构与八条设计建议
| 架构 | 谁组织下一步 | 本章联系 |
|---|---|---|
| 传统软件 | 预先编写的分支 | 日期、状态、金额比较直接计算 |
| LLM 智能体 | 模型参与选择动作与工具 | 开放任务中可用,但需要监督与约束 |
| AI 驱动的软件 | 代码组织,模型提供狭窄判断 | 下面的客服工作流 |
依次实践:能用代码就用代码;只送相关上下文;给输入命名;分解问题; 必要时结构化 instructions 和 criteria;批量提出独立问题;代码组合;按不确定性路由。
结构化、并行、可比较是接口与架构能力。重复稳定性、具体速度和中文表现需要实测,不能用文档的宣传性数字代替本项目记录。
先运行确定性规则
复刻逾期发票例子:日期差不用询问模型。只返回标签,不触发实际催收。
from datetime import date
TODAY = date(2026, 9, 22)
INVOICE_DUE = date(2026, 8, 1)
days_overdue = (TODAY - INVOICE_DUE).days
invoice_route = "collections_review" if days_overdue > 30 else "normal"显示确定性结果。
print({"逾期天数": days_overdue, "建议路径": invoice_route}){'逾期天数': 52, '建议路径': 'collections_review'}观察与理解: 若真实系统已有规则,应直接调用规则。这里固定日期是为了可复现,不表示程序获取了当前业务日期。
2. 复刻完整客服分流
原理与理论根基
沿用原文的 billing、orders、account 三类和七个问题:主题、索要凭据、发件身份不匹配、意外奖励、退款请求、开放订单引用、不满程度。 垃圾信息风险由三个 Noul 在代码中加权得到;它是应用分数,不是已校准的垃圾概率。 只有当前路径用得上的答案才参与后续处理,例如 billing 使用退款信号,orders 使用订单引用信号。
第一步:准备客户资料
订单状态是系统枚举;模型只收到仍未送达的订单。
CUSTOMER = {
"plan": "团队版",
"orders": [
{"id": "A-104", "status": "processing", "description": "办公用品订单"},
{"id": "A-090", "status": "delivered", "description": "已送达的历史订单"},
],
}
STANDARD_SENDER = {"display_name": "客户王小明", "email": "xiaoming@customer.example"}定义正常业务消息。
TICKETS = [
{"id": "closed", "status": "closed", "message": "退款问题已经解决。"},
{"id": "billing", "status": "open", "message": "我被重复扣款,请退还多扣的金额。"},
{"id": "orders", "status": "open", "message": "请问我的订单 A-104 什么时候发货?"},
{"id": "account_normal", "status": "open", "message": "请告诉我怎样重置账户密码。"},
{"id": "account_high", "status": "open", "message": "账号又无法登录!投诉多次没人处理,再不解决我就注销账户!"},
]增加原文凭据与意外奖励场景,以及用于寻找复核分支的模糊消息。
TICKETS += [
{"id": "quarantine", "status": "open",
"sender": {"display_name": "Acme 薪资部门", "email": "rewards@claim-bonus.example"},
"message": "恭喜您获得意外奖金!请今天回复您的薪资账户密码领取。"},
{"id": "review_spam", "status": "open",
"sender": {"display_name": "薪资服务", "email": "notice@service.example"},
"message": "你可能有一笔奖励待确认,请尽快处理账户验证。"},
{"id": "review_topic", "status": "open", "message": "那个问题还是没好,行,就按你说的办。"},
]观察与理解: id 是记录标识,不是会发送的标准答案;真实模型可能把模糊消息判得很确定,也可能走其他路径。
用白名单构造相关上下文。
def build_ticket_state(ticket, customer):
return {
"ticket": {
"message": ticket["message"],
"sender": ticket.get("sender", STANDARD_SENDER),
"links": ticket.get("links", []),
},
"customer": {
"plan": customer["plan"],
"open_orders": [x for x in customer["orders"] if x["status"] != "delivered"],
},
"policy": {"sensitive_credentials": ["密码", "安全验证码", "API 密钥"]},
}第二步:原子问题与结构化定义
criteria 采用相同字段说明涵盖内容和排除项,减少账单、订单和账户主题混淆。模型看的是完整描述。
先定义主题分类。
WORKFLOW_QUESTIONS = {
"topic": Choice(
instructions={"question": "哪个团队应处理 `ticket.message`?", "focus": "只按主要诉求分类。"},
criteria={
"billing": {"what": "扣款、发票、退款或订阅", "not_for": "订单跟踪或登录"},
"orders": {"what": "订单状态、配送、取消或退货", "not_for": "账单或登录"},
"account": {"what": "登录、个人资料、权限或账户安全", "not_for": "扣款或配送"},
},
),
}加入凭据请求:使用 NoulCriteria 区分索要密码与指导重置密码。
WORKFLOW_QUESTIONS["requests_credentials"] = Noul(
instructions={"question": "消息是否要求披露敏感凭据本身?",
"compare": ["`ticket.message`", "`policy.sensitive_credentials`"]},
criteria=NoulCriteria(
true={"what": "要求收件人提供列出的敏感凭据", "examples": ["请回复你的密码"]},
false={"what": "没有要求披露凭据", "examples": ["请使用重置链接修改密码"]},
),
)观察与理解: 不要把出现‘密码’两个字一律当作索要凭据。这正是狭窄语义判断比关键词规则更值得试验的地方。
加入其他相互独立的 Noul。
WORKFLOW_QUESTIONS.update({
"sender_identity_mismatch": Noul(instructions=(
"`ticket.sender.display_name` 声称的机构与 `ticket.sender.email` 的域名是否明显冲突?")),
"unexpected_reward": Noul(instructions=(
"`ticket.message` 是否宣称收件人获得未申请的意外奖金或奖品?已知退款不算。")),
"refund_requested": Noul(instructions=(
"客户在 `ticket.message` 中是否明确要求退款或账户抵扣?仅抱怨扣款不算。")),
"mentions_open_order": Noul(instructions=(
"`ticket.message` 是否用编号或可辨认细节提到了 `customer.open_orders` 中的订单?")),
})最后加入不满程度量表。
WORKFLOW_QUESTIONS["frustration"] = Score(
instructions="客户在 `ticket.message` 中表达了多大程度的不满?判断情绪,不判断故障严重程度。",
criteria=["平静且客观", "不满但保持礼貌", "非常愤怒或威胁离开"],
)观察与理解: 所有问题看到同一份 state。哪个答案参与决策由后面的代码决定。
第三步:在代码中组合
沿用参考示例的权重与阈值以便对照;这些不是本项目验证出的最佳值。真实应用需要独立标签来选择阈值。
先写风险分数组合。
def spam_score(response):
return (
0.45 * response.nouls["requests_credentials"].noul
+ 0.30 * response.nouls["sender_identity_mismatch"].noul
+ 0.25 * response.nouls["unexpected_reward"].noul
)再写有限且可观察的路由。
def decide_route(response):
topic = response.choices["topic"]
risk = spam_score(response)
if 0.4 < risk < 0.6:
return {"route": "review_spam", "risk": risk}
if topic.confidence < 0.75:
return {"route": "review_topic", "risk": risk}
if risk >= 0.6:
return {"route": "quarantine", "risk": risk}
if topic.choice == "billing":
return {"route": "billing", "refund_requested": response.nouls["refund_requested"].noul >= 0.7}
if topic.choice == "orders":
return {"route": "orders", "mentions_open_order": response.nouls["mentions_open_order"].noul >= 0.7}
frustration = response.scores["frustration"]
high = frustration.confidence >= 0.7 and frustration.score >= 1.5
return {"route": "account_high" if high else "account_normal"}观察与理解: 退款信号只传给账单队列,不能触发真实退款。网络错误会停止实验;本章不把失败请求算成模型低置信度。
已关闭的工单先跳过 API,其余工单构造状态并请求。
def triage_ticket(ticket):
if ticket["status"] == "closed":
return {"route": "no_action"}, None
state = build_ticket_state(ticket, CUSTOMER)
response = ts.call(state, WORKFLOW_QUESTIONS, WORKFLOW_OFFLINE[ticket["id"]], ticket["id"])
return decide_route(response), response3. 换几种说法试模型并核对覆盖
这里选择可读的消息观察行为,而不是把这八条当作准确率基准。用于修改提示词的样例属于开发数据。
运行八个案例;closed 不调用 API,因此发起七次业务请求。
workflow_results = {ticket["id"]: triage_ticket(ticket) for ticket in TICKETS}逐条显示路由和模型信号,保留与预想不一致的结果。
for case_id, (decision, response) in workflow_results.items():
print("案例:", case_id, "结果:", decision)
if response is not None:
show(response)案例: closed 结果: {'route': 'no_action'}
案例: billing 结果: {'route': 'billing', 'refund_requested': True}
{
"topic": {
"type": "choice",
"choice": "billing",
"confidence": 1.0,
"probabilities": {
"orders": 0.0,
"billing": 1.0,
"account": 0.0
}
},
"requests_credentials": {
"type": "noul",
"noul": 0.03
},
"sender_identity_mismatch": {
"type": "noul",
"noul": 0.08
},
"unexpected_reward": {
"type": "noul",
"noul": 0.01
},
"refund_requested": {
"type": "noul",
"noul": 0.96
},
"mentions_open_order": {
"type": "noul",
"noul": 0.02
},
"frustration": {
"type": "score",
"score": 0.84,
"confidence": 0.75,
"probabilities": {
"0": 0.16,
"1": 0.84,
"2": 0.0
},
"legend": {
"0": "平静且客观",
"1": "不满但保持礼貌",
"2": "非常愤怒或威胁离开"
}
}
}
案例: orders 结果: {'route': 'orders', 'mentions_open_order': True}
{
"topic": {
"type": "choice",
"choice": "orders",
"confidence": 1.0,
"probabilities": {
"orders": 1.0,
"billing": 0.0,
"account": 0.0
}
},
"requests_credentials": {
"type": "noul",
"noul": 0.02
},
"sender_identity_mismatch": {
"type": "noul",
"noul": 0.08
},
"unexpected_reward": {
"type": "noul",
"noul": 0.01
},
"refund_requested": {
"type": "noul",
"noul": 0.01
},
"mentions_open_order": {
"type": "noul",
"noul": 0.98
},
"frustration": {
"type": "score",
"score": 0.01,
"confidence": 0.99,
"probabilities": {
"0": 0.99,
...(输出过长,已截断)观察与理解: 离线模式覆盖所有分支是因为数据被人工设计过。live 运行若缺少复核分支,必须报告未观察到。
计算真实观察到的路径集合。
EXPECTED_ROUTES = {
"no_action", "review_spam", "review_topic", "quarantine",
"billing", "orders", "account_normal", "account_high",
}
observed_routes = {decision["route"] for decision, _ in workflow_results.values()}
COVERAGE = {
"expected": sorted(EXPECTED_ROUTES), "observed": sorted(observed_routes),
"missing": sorted(EXPECTED_ROUTES - observed_routes),
"source": "live" if all(x["source"] == "live" for x in CALL_LOG) else "offline",
}显示覆盖报告。
print(json.dumps(COVERAGE, ensure_ascii=False, indent=2)){
"expected": [
"account_high",
"account_normal",
"billing",
"no_action",
"orders",
"quarantine",
"review_spam",
"review_topic"
],
"observed": [
"account_high",
"account_normal",
"billing",
"no_action",
"orders",
"quarantine"
],
"missing": [
"review_spam",
"review_topic"
],
"source": "live"
}观察与理解: 缺失路径需要补几句有记录的试句或如实保留缺口;调整阈值需说明理由,不能把离线输出混进 live 补齐。
4. 原文其他设计建议的小配方
下面复刻航班退款、嵌套路径、候选记录去重、虚拟卡分类与天气工具轨迹五个示例。 每个只做一次请求。它们帮助理解输入和问题的设计,不展开其他作者负责的完整架构模式。
4.1 只送相关上下文:航班退款
FLIGHT = {
"ticket_message": "我的航班取消了,可以退款吗?",
"refund_policy": "取消的航班可申请全额退款。",
}问题明确要求按给定政策判断。
FLIGHT_QUESTIONS = {
"policy_supports_refund": Noul(instructions="提供的 refund_policy 是否支持 ticket_message 中请求的退款?"),
}调用。
flight_response = ts.call(FLIGHT, FLIGHT_QUESTIONS, FLIGHT_OFFLINE, "航班退款政策")观察。
show(flight_response){
"policy_supports_refund": {
"type": "noul",
"noul": 0.97
}
}观察与理解: 给定政策是本例事实;不能凭模型记忆替代最新的公司政策。
4.2 嵌套输入:明确引用不同消息与记录
NESTED_STATE = {
"support": {"tickets": [
{"message": "订单 A-104 被扣了两次款。"},
{"message": "怎样重置我的密码?"},
]},
"commerce": {"orders": [{"id": "A-104", "charges": [
{"amount_usd": 49, "status": "captured"},
{"amount_usd": 49, "status": "captured"},
]}]},
"account": {"security": {"password_reset": "向已登记邮箱发送密码重置链接。"}},
}用反引号路径消除引用歧义。
NESTED_QUESTIONS = {
"duplicate_charge": Noul(instructions=(
"`support.tickets[0].message` 与 `commerce.orders[0].charges` 是否表明重复扣款?")),
"password_reset_supported": Noul(instructions=(
"`account.security.password_reset` 能否解决 `support.tickets[1].message` 中的请求?")),
}调用。
nested_response = ts.call(NESTED_STATE, NESTED_QUESTIONS, NESTED_OFFLINE, "嵌套状态路径")观察。
show(nested_response){
"duplicate_charge": {
"type": "noul",
"noul": 0.96
},
"password_reset_supported": {
"type": "noul",
"noul": 0.94
}
}观察与理解: 反引号路径是给模型的文字指引,不是客户端执行的数据库查询。
4.3 把代码中的候选记录放进 instructions
RESUME = {"resume": {
"name": "约翰·史密斯", "location": "加利福尼亚州奥克兰",
"summary": "有八年 Python 和 Go 经验的后端工程师",
"experience": [
{"employer": "Google", "title": "高级后端工程师", "years": "2021-2025"},
{"employer": "Microsoft", "title": "软件工程师", "years": "2017-2021"},
],
}}候选记录独立具名,不需要拼接长提示词。
RECORD_QUESTIONS = {
"same_as_record_18": Noul(instructions={
"potential_duplicate": {"name": "约翰·史密斯", "location": "加州奥克兰", "last_employer": "Google"},
"question": "简历和 potential_duplicate 是否很可能描述同一个人?",
}),
}调用。
record_response = ts.call(RESUME, RECORD_QUESTIONS, RECORD_OFFLINE, "候选记录比较")观察。
show(record_response){
"same_as_record_18": {
"type": "noul",
"noul": 0.94
}
}观察与理解: 返回的是匹配信号,不能单凭同名自动合并真实人员档案。
4.4 对比式 Choice criteria:虚拟卡的申请与限制
CARD_MESSAGE = "我每天最多能创建多少张一次性虚拟卡?"每个选项使用一致字段说明涵盖与排除范围。
CARD_QUESTIONS = {
"card_help_topic": Choice(
instructions={"question": "用户询问哪种一次性虚拟卡主题?", "focus": "按用户想获得的信息分类。"},
criteria={
"get_disposable_virtual_card": {
"what": "用途、资格或开通方式", "not_for": "数量、交易或商户限制",
"examples": ["怎样申请一次性虚拟卡?", "一次性卡有什么用途?"],
},
"disposable_card_limits": {
"what": "数量、交易或商户限制", "not_for": "用途、资格或开通方式",
"examples": ["每天可以创建多少张?", "哪些商户可以使用?"],
},
},
),
}调用。
card_response = ts.call(CARD_MESSAGE, CARD_QUESTIONS, CARD_OFFLINE, "虚拟卡主题")观察。
show(card_response){
"card_help_topic": {
"type": "choice",
"choice": "disposable_card_limits",
"confidence": 1.0,
"probabilities": {
"disposable_card_limits": 1.0,
"get_disposable_virtual_card": 0.0
}
}
}观察与理解: 标签应定位到数量限制;这是待验证预期,不是对任何一次请求的强制断言。
4.5 分解工具调用轨迹
复刻原文的西雅图天气查询。请求要华氏度,轨迹却使用摄氏度。这里只检查提供的轨迹,没有调用真实天气工具。
准备用户请求与工具定义。
TRACE_STATE = {
"request": {"text": "请查西雅图 2026 年 9 月 3 日的天气,使用华氏度。",
"location": "西雅图", "date": "2026-09-03", "unit": "fahrenheit"},
"available_tools": {
"geocode_city": {"description": "把城市解析为经纬度", "parameters": {"city": "string"}},
"get_weather": {"description": "查询指定日期和坐标的天气", "parameters": {
"latitude": "number", "longitude": "number", "date": "YYYY-MM-DD",
"unit": ["fahrenheit", "celsius"],
}},
},
}加入已记录的调用轨迹。
TRACE_STATE["trace"] = {
"tool_calls": [
{"id": "call_1", "name": "geocode_city", "arguments": {"city": "西雅图"}},
{"id": "call_2", "name": "get_weather", "arguments": {
"latitude": 47.6062, "longitude": -122.3321,
"date": "2026-09-03", "unit": "celsius",
}},
],
"tool_results": [{"tool_call_id": "call_1", "output": {"latitude": 47.6062, "longitude": -122.3321}}],
}九个独立命题定位各类错误。
TRACE_PROMPTS = {
"geocode_tool_is_relevant": "第一个工具是否适合解析 request.location 的地理坐标?",
"geocode_location_matches": "第一个调用的城市是否与 request.location 匹配?",
"geocode_arguments_match_schema": "第一个调用参数是否符合 geocode_city 参数定义?",
"geocode_result_matches_call": "tool_results[0].tool_call_id 是否与第一个调用的 id 匹配?",
"weather_tool_is_relevant": "第二个工具是否适合回答 request.text 的天气查询?",
"weather_arguments_match_schema": "第二个调用参数是否符合 get_weather 参数定义?",
"weather_uses_geocoded_coordinates": "第二个调用坐标是否与第一个工具结果匹配?",
"weather_date_matches": "第二个调用日期是否与 request.date 匹配?",
"weather_unit_matches": "第二个调用单位是否与 request.unit 匹配?",
}
TRACE_QUESTIONS = {key: Noul(instructions=value) for key, value in TRACE_PROMPTS.items()}观察与理解: 精确的日期、枚举和 schema 校验可以直接用代码。这里保留原文的原子判断演示,帮助定位宽泛‘轨迹正确吗’背后的因素。
调用。
trace_response = ts.call(TRACE_STATE, TRACE_QUESTIONS, TRACE_OFFLINE, "工具轨迹九问")观察各项,再用确定性规则确认单位。
show(trace_response){
"geocode_tool_is_relevant": {
"type": "noul",
"noul": 0.97
},
"geocode_location_matches": {
"type": "noul",
"noul": 0.99
},
"geocode_arguments_match_schema": {
"type": "noul",
"noul": 0.98
},
"geocode_result_matches_call": {
"type": "noul",
"noul": 0.96
},
"weather_tool_is_relevant": {
"type": "noul",
"noul": 0.41
},
"weather_arguments_match_schema": {
"type": "noul",
"noul": 0.62
},
"weather_uses_geocoded_coordinates": {
"type": "noul",
"noul": 0.98
},
"weather_date_matches": {
"type": "noul",
"noul": 0.98
},
"weather_unit_matches": {
"type": "noul",
"noul": 0.02
}
}观察与理解: 预期单位一致性的概率低。如果模型漏检,记录漏检;不让模型覆盖已知的确定性事实。
补上可以精确执行的检查。
unit_matches = TRACE_STATE["trace"]["tool_calls"][1]["arguments"]["unit"] == TRACE_STATE["request"]["unit"]
print({"代码检查单位一致": unit_matches})
assert unit_matches is False{'代码检查单位一致': False}观察与理解: 这是确定性数据中的已知错误,因此可以断言。业务验收仍要分别评价其他语义检查。
练习与自查
查看 COVERAGE:哪些路径实际没有命中?为一条缺失路径提出新措辞,先写假设、再运行并记录。另想一个测试集应该如何独立于这些试句。
参考思路:先完成练习再展开
保留所有尝试及真实返回,不只保留命中的样例。调过问题或阈值后,用另一批未参与调整的标注工单评估覆盖率与误分;不得强改 API 数值填补路径。
小结
| 设计动作 | 本章实现 |
|---|---|
| 保留代码规则 | 逾期、closed、单位比较 |
| 狭窄语义判断 | 七问分流、九问轨迹 |
| 结构化定义 | 候选记录、虚拟卡 criteria |
| 覆盖核对 | 自动列出 observed / missing |
下一章:AI 入门。更完整的扇出、门控与组合评分见架构模式章。
离线运行只说明教材代码能执行。正式交付必须实际运行 live,并阅读每条输出;缺失的分支应记为未观察到。
本次执行记录
先关闭连接,再生成记录。下面的 JSON 由实际运行计算,批量执行器会据此检查来源。
if client is not None:
client.close()本章拿几句话试了试真实模型,看它给的概率怎么反应——这只说明模型对这类输入的反应方式,不构成准确率评测。特别提醒:如果哪句答得合心意就专门挑出来当考题,再拿这些挑过的句子去算准确率,数字必然虚高。输出里的耗时也只是当时网络的快照,每次都会不一样。
AUDIT = {
"kind": "jev_execution_audit",
"executed_at_utc": datetime.now(timezone.utc).isoformat(),
"sdk": version("typesafe-sdk"), "requested_model": MODEL,
"mode": RUN_MODE, "ping": PING,
"real_calls": sum(x["source"] == "live" for x in CALL_LOG),
"offline_calls": sum(x["source"] == "offline" for x in CALL_LOG),
"cases": CALL_LOG,
"coverage": globals().get("COVERAGE", {}),
"validation_status": "live_executed_requires_review" if (
PING["source"] == "live" and CALL_LOG
and all(x["source"] == "live" for x in CALL_LOG)
) else "offline_only_not_model_evidence",
}
print(json.dumps(AUDIT, ensure_ascii=False, indent=2)){
"kind": "jev_execution_audit",
"executed_at_utc": "2026-09-25T09:08:26.673030+00:00",
"sdk": "0.7.1",
"requested_model": "jev-1.13.0",
"mode": "auto",
"ping": {
"source": "live",
"model": "jev-1.13.0",
"input_tokens": 278,
"output_tokens": 22
},
"real_calls": 12,
"offline_calls": 0,
"cases": [
{
"case": "billing",
"source": "live",
"model": "jev-1.13.0",
"seconds": 0.5380647080019116,
"input_tokens": 979,
"output_tokens": 144
},
{
"case": "orders",
"source": "live",
"model": "jev-1.13.0",
"seconds": 0.3027524589560926,
"input_tokens": 982,
"output_tokens": 144
},
{
"case": "account_normal",
"source": "live",
"model": "jev-1.13.0",
"seconds": 0.2705029579810798,
"input_tokens": 976,
"output_tokens": 144
},
{
"case": "account_high",
"source": "live",
"model": "jev-1.13.0",
"seconds": 0.28899812512099743,
"input_tokens": 991,
"output_tokens": 144
},
{
"case": "quarantine",
"source": "live",
"model": "jev-1.13.0",
"seconds": 0.2741255830042064,
"input_tokens": 994,
"output_tokens": 144
},
{
"case": "review_spam",
"source": "live",
"model": "jev-1.13.0",
"seconds": 0.28349100006744266,
"input_tokens": 982,
"output_tokens": 144
},
{
"case": "review_topic",
"source": "live",
"model": "jev-1.13.0",
"seconds": 0.28805779200047255,
"input_tokens": 981,
...(输出过长,已截断)读完输出后,在本仓库 notebooks/MAINTENANCE.md 的验收表中记录日期、真实模型、观察到的分支和偏离预期之处。不要把人工演示数值抄进实测记录。