chenlin
0d986f60bd
web-platform-ci / Standalone decision service (no cloud credentials) (push) Has been cancelled
web-platform-ci / TypeScript, lint, unit, build (push) Has been cancelled
web-platform-ci / Playwright E2E (push) Has been cancelled
lekiwi-compatibility / cpu-compatibility (push) Has been cancelled
集成服务器托管模型、自然语言移动与有界抓放、内置 LeKiwi URL 导入和双摄像头；同步部署契约与指定域名 iframe 白名单，保留原有物理安全、会话及调用预算防护。 更新 npm 包及锁文件版本、CHANGELOG 与发布文档。提交前 typecheck、120 项定向前端测试和 44 项后端测试通过（3 项可选跳过）；真实 v2 云模型抓放仍待单独验收，不包含运行密钥或构建产物。
111 lines
4.4 KiB
Python
111 lines
4.4 KiB
Python
"""TypeSafe System One / OpenRouter Decisions protocol, not a chat endpoint.
|
|
|
|
Protocol shape informed by MIT-licensed jev-libero / embodied-jev; see THIRD_PARTY_NOTICES.
|
|
"""
|
|
|
|
import json
|
|
import math
|
|
|
|
from ..protocol import LANGUAGE_VERSION, DecisionError, schema_for, validate_jev
|
|
from .http import post, usage
|
|
|
|
|
|
def question(options, instruction):
|
|
return {
|
|
"type": "choice",
|
|
"instructions": instruction,
|
|
"criteria": {option: option for option in options},
|
|
}
|
|
|
|
|
|
def answer(result, name, options):
|
|
answers = result.get("answers")
|
|
item = answers.get(name) if isinstance(answers, dict) else None
|
|
if not isinstance(item, dict) or item.get("choice") not in options:
|
|
raise DecisionError("jev_invalid_choice", 502)
|
|
probabilities = item.get("probabilities", {})
|
|
if (
|
|
not isinstance(probabilities, dict)
|
|
or probabilities.keys() - set(options)
|
|
or any(
|
|
type(p) not in (int, float) or not math.isfinite(p) or not 0 <= p <= 1
|
|
for p in probabilities.values()
|
|
)
|
|
):
|
|
raise DecisionError("jev_invalid_probabilities", 502)
|
|
# Keep official choice even when probabilities do not rank it highest.
|
|
return item["choice"]
|
|
|
|
|
|
async def decide(session, conn, request):
|
|
version = request["observation"]["version"]
|
|
props = schema_for(version)["$defs"]["JevDecision"]["properties"]
|
|
options = {name: props[name]["enum"] for name in ("grasp", "diagnosis", "recovery")}
|
|
options["choice"] = request["candidates"]
|
|
prompts = {
|
|
"choice": (
|
|
"Choose an offered next skill if safe. Stop for hard safety faults. "
|
|
"Only the local controller can determine success or permit actuation."
|
|
),
|
|
"grasp": (
|
|
"secure requires two finger forces >=0.2 N, verified lift and stable grasp evidence. "
|
|
"empty means no held object. slipping means a previously secure grasp is being lost. "
|
|
"Use uncertain if evidence is insufficient. An open gripper is not a secure grasp."
|
|
),
|
|
"diagnosis": (
|
|
"Report none when failure=none and there are no safety flags. "
|
|
"An empty gripper before closing or after release is expected, not a fault. "
|
|
"Otherwise diagnose empty, slipping, misaligned, unreachable, stalled or uncertain."
|
|
),
|
|
"recovery": (
|
|
"Continue with a safe next skill if failure=none; retry only recoverable alignment "
|
|
"or empty grasp failures before transport; replan when retries are insufficient; "
|
|
"stop for hard safety faults or unsafe uncertainty."
|
|
),
|
|
}
|
|
if version == LANGUAGE_VERSION:
|
|
options.update({name: props[name]["enum"] for name in ("noul", "score", "reason")})
|
|
prompts.update(
|
|
{
|
|
"noul": (
|
|
"Veto unsafe actions: allow only with fresh sufficient evidence and no hard "
|
|
"safety fault; deny on danger, uncertain on missing evidence. "
|
|
"This is not physical authorization."
|
|
),
|
|
"score": (
|
|
"Grade progress at this skill boundary: poor, partial, good, excellent, "
|
|
"or unavailable. This is a discrete quality grade, NOT probability or "
|
|
"final success. Empty fingers before grasp/after release are expected."
|
|
),
|
|
"reason": (
|
|
"Select the main reason for safety/quality: none, unsafe, "
|
|
"insufficient_evidence, tracking_error, verified_progress."
|
|
),
|
|
}
|
|
)
|
|
result = await post(
|
|
session,
|
|
conn,
|
|
"",
|
|
{
|
|
"model": conn.model,
|
|
**(
|
|
{"provider": {"allow_fallbacks": False}}
|
|
if conn.protocol == "openrouter-decisions"
|
|
else {}
|
|
),
|
|
"state": json.dumps(
|
|
{"observation": request["observation"], "failure": request.get("failure", "none")},
|
|
ensure_ascii=False,
|
|
),
|
|
"questions": {
|
|
name: question(values, prompts[name]) for name, values in options.items()
|
|
},
|
|
},
|
|
)
|
|
value = {
|
|
"version": version,
|
|
**{name: answer(result, name, values) for name, values in options.items()},
|
|
}
|
|
return validate_jev(value, request["candidates"], version), usage(result)
|