128 lines
6.4 KiB
Python
128 lines
6.4 KiB
Python
"""确定性优先的任务规划；仅在无法分类页面时才请求可选 LLM。"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import base64
|
||
import urllib.request
|
||
from dataclasses import dataclass
|
||
|
||
from .tools import REGISTRY
|
||
|
||
|
||
class PlanError(RuntimeError):
|
||
pass
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Step:
|
||
tool: str
|
||
arguments: dict
|
||
reason: str
|
||
|
||
|
||
def classify(texts: list[str]) -> str:
|
||
content = "\n".join(texts)
|
||
if "证件号码后8位" in content or "请输入乘车人" in content:
|
||
return "verify"
|
||
if "发票信息确认" in content:
|
||
return "invoice_confirm"
|
||
if "发票开具成功" in content:
|
||
return "success"
|
||
if "发票信息" in content and ("提交" in content or "确认" in content):
|
||
return "invoice_info"
|
||
if "没有可开具电子发票" in content:
|
||
return "no_ticket"
|
||
if "开具" in content and "*" in content:
|
||
return "list"
|
||
return "unknown"
|
||
|
||
|
||
def deterministic_plan(task: dict) -> list[Step]:
|
||
history = task.get("history", [])
|
||
navigation = task.get("navigation", {})
|
||
phase = navigation.get("phase")
|
||
if phase in {"push_qr", "open_scan", "open_orders", "open_album", "select_qr", "wait_list"}:
|
||
if phase not in {"push_qr", "wait_list"} and not task.get("allow_submit"):
|
||
return []
|
||
return [Step(f"nav.{phase}", {}, "进入扫码开票单列表")]
|
||
if task.get("invoice", {}).get("finished"):
|
||
return []
|
||
if not history:
|
||
return [Step("device.status", {}, "确认唯一目标设备"), Step("observe.screen", {}, "保存当前屏幕"), Step("observe.xml", {}, "读取可访问 UI 文本")]
|
||
if history[-1].get("tool") not in {"observe.xml", "invoice.inspect"}:
|
||
return [Step("observe.screen", {}, "动作后保存当前屏幕"), Step("observe.xml", {}, "动作后重新读取 UI 文本")]
|
||
last = next((entry for entry in reversed(history) if entry.get("tool") == "observe.xml"), None)
|
||
page = classify(last.get("result", {}).get("texts", [])) if last else "other"
|
||
if page in {"other", "unknown"} and history[-1].get("tool") == "invoice.inspect":
|
||
page = str(history[-1].get("result", {}).get("page", "other"))
|
||
if page in {"other", "unknown"}:
|
||
return []
|
||
if page in {"other", "unknown"}:
|
||
return [Step("invoice.inspect", {}, "XML 不足时由视觉识别页面")]
|
||
if page == "loading":
|
||
return [Step("invoice.wait", {"seconds": 2}, "等待开票网络请求完成")]
|
||
if not task.get("allow_submit"):
|
||
return []
|
||
routes = {
|
||
"list": ("invoice.open_next", "匹配名单并打开下一位乘客"),
|
||
"verify": ("invoice.verify", "填写证件后八位并核验"),
|
||
"invoice_info": ("invoice.submit_info", "提交发票信息"),
|
||
"invoice_confirm": ("invoice.confirm", "确认开具"),
|
||
"no_ticket": ("invoice.no_ticket", "记录无票并返回列表"),
|
||
}
|
||
if page == "success":
|
||
action = task.get("invoice", {}).get("success_action", "download")
|
||
return [Step("invoice.email", {}, "发送发票邮件并继续开票")] if action == "email" else [Step("invoice.download", {}, "下载发票、归档并继续开票")]
|
||
if page in routes:
|
||
tool, reason = routes[page]
|
||
return [Step(tool, {}, reason)]
|
||
return []
|
||
|
||
|
||
def validate_llm_steps(raw: object, *, allow_submit: bool) -> list[Step]:
|
||
if not isinstance(raw, list):
|
||
raise PlanError("LLM 返回必须是 steps 数组")
|
||
if len(raw) > 3:
|
||
raise PlanError("LLM 单次最多重规划 3 个原子工具")
|
||
steps: list[Step] = []
|
||
for item in raw:
|
||
if not isinstance(item, dict):
|
||
raise PlanError("LLM step 必须是对象")
|
||
tool = item.get("tool")
|
||
if tool not in REGISTRY:
|
||
raise PlanError(f"LLM 请求了未注册工具：{tool}")
|
||
if tool in {"ui.tap_normalized", "invoice.open_next", "invoice.verify", "invoice.submit_info", "invoice.confirm", "invoice.download", "invoice.email", "invoice.no_ticket", "nav.open_scan", "nav.open_orders", "nav.open_album", "nav.select_qr"} and not allow_submit:
|
||
raise PlanError("默认禁止 LLM 直接点击业务按钮；请先在任务中显式启用 allow_submit")
|
||
arguments = item.get("arguments", {})
|
||
if not isinstance(arguments, dict):
|
||
raise PlanError("工具参数必须是对象")
|
||
steps.append(Step(tool, arguments, str(item.get("reason") or "LLM 重规划")))
|
||
return steps
|
||
|
||
|
||
def llm_plan(task: dict, settings: dict) -> list[Step]:
|
||
planner = settings.get("planner", {})
|
||
if not planner.get("enabled"):
|
||
return []
|
||
base_url, model = str(planner.get("base_url", "")).rstrip("/"), str(planner.get("model", ""))
|
||
api_key = os.environ.get(str(planner.get("api_key_env", "AUTOTRAIN_NEXT_LLM_API_KEY")), "")
|
||
if not (base_url and model and api_key):
|
||
raise PlanError("LLM 已启用但 base_url、model 或 API key 环境变量缺失")
|
||
observations = [entry.get("result", {}) for entry in task.get("history", []) if entry.get("tool") == "observe.screen"]
|
||
safe_task = {"type": task.get("type"), "allow_submit": bool(task.get("allow_submit")), "page": task.get("invoice", {}).get("page"), "recent_tools": [entry.get("tool") for entry in task.get("history", [])[-6:]]}
|
||
content: list[dict] = [{"type": "text", "text": json.dumps({"task": safe_task, "tools": list(REGISTRY)}, ensure_ascii=False)}]
|
||
if observations and observations[-1].get("screenshot"):
|
||
try:
|
||
image = open(observations[-1]["screenshot"], "rb").read()
|
||
content.append({"type": "image_url", "image_url": {"url": "data:image/png;base64," + base64.b64encode(image).decode()}})
|
||
except OSError:
|
||
pass
|
||
payload = {"model": model, "temperature": 0, "messages": [{"role": "system", "content": "你是手机自动化规划器。只返回 JSON：{\\\"steps\\\":[{\\\"tool\\\":...,\\\"arguments\\\":{},\\\"reason\\\":...}]}。不可编造工具。"}, {"role": "user", "content": content}]}
|
||
request = urllib.request.Request(base_url + "/chat/completions", data=json.dumps(payload).encode(), headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"})
|
||
with urllib.request.urlopen(request, timeout=45) as response:
|
||
body = json.loads(response.read())
|
||
content = body["choices"][0]["message"]["content"]
|
||
parsed = json.loads(content)
|
||
return validate_llm_steps(parsed.get("steps"), allow_submit=bool(task.get("allow_submit")))
|