1. ELG 视线估计从本地脚本到服务化,卡在哪一步
ELG(Eye Region Landmarks based Gaze Estimation)是一套基于眼区标志点的视线估计方法,核心思路是先稳定地找到眼睛周围的 18 个关键点,再用这些点去拟合虹膜中心和视线方向。它和纯外观回归的 Gaze Estimation 模型相比,优势在于对光照、头部姿态的鲁棒性更好,适合做远程视线追踪、注意力分析、驾驶疲劳监测这类场景。如果你正在看 Gaze Estimation 教程,大概率已经跑通了 GazeML 的vid_demo.py,能在一段视频上画出眼睛关键点和虹膜圈。
但真正要把它用起来,问题往往不在模型本身,而在“怎么让别的程序调用它”。本地脚本是单进程、单机、写死路径的:模型文件放在outputs/或MODELS_DIR,输入是本地视频文件,输出直接cv2.imshow。一旦你想让前端上传一张图、让另一个服务批量跑推理、或者让移动端拿到视线结果,就得自己搭一套 HTTP 服务。这时候会撞上三件事:模型加载和会话管理、请求参数校验、以及最烦的——多个模型/多个服务之间的鉴权与 Key 管理。
我试过把 ELG 的 PB 模型直接塞进 Flask,每个请求tf.Session()重建一次,QPS 低得可怜;也试过把 Key 硬编码在客户端,结果换环境就要重新打包。这篇就聚焦 ELG 视线估计模型从本地推理到服务化调用的工程落地,给出config.toml与settings.json的可复制骨架,演示通过 TaoToken 统一 Key/API 通道接入推理服务,并附一次请求验证与返回字段核对动作。目标很明确:你照着改完,能用一个 Key 同时管住模型推理服务和后续要接的对话/编码类能力,不用在每台机器上散落一堆密钥。
2. 前置准备:TaoToken 统一 Key 与 ELG 推理服务的关系
先把定位说清楚。TaoToken 在这里扮演的是“统一入口”的角色:你不需要为每个下游能力单独申请一套凭证,而是拿一个 Key,通过统一的 API 通道去访问。对 ELG 这种自建推理服务来说,最实用的做法是——把 ELG 服务本身当成一个受保护的后端,前面用 TaoToken 的 Key 做鉴权层,客户端只认这一个 Key。
你需要准备的东西不多:
- 一个 TaoToken 账号,登录后进入控制台创建 API Key。地址是
https://taotoken.net/api,控制台里可以管理 Key 和额度。 - 本地已经能跑通 ELG 的推理代码,至少能加载
gaze_opt_b2.m.pb或gaze-faceX-lms5.pb,输入2x108x180x1的眼睛批次,拿到landmarks、radius两个输出。 - Python 环境,建议 3.8+,装好
tensorflow(1.x 或 2.x 兼容模式都行,看你的 PB 是哪个版本导出的)、opencv-python、flask、requests、toml。
关于 Key 的获取,直接去控制台的 API Keys 页面新建即可,建议按环境分 Key,比如elg-dev、elg-prod,方便后面排查是哪个环境在打请求。文档在https://taotoken.net/api下有接入说明,遇到字段对不上时优先翻文档而不是猜。
注意:Key 只放在服务端,不要写进前端 JS 或移动端包体。客户端拿到的应该是你自己服务的临时凭证,而不是 TaoToken 的 Key。
3. 可复制配置:config.toml 与 settings.json 骨架
工程落地的第一步是把“会变的东西”抽出来。ELG 推理服务里会变的主要是:模型路径、输入尺寸、batch、监听端口、以及 TaoToken 的接入地址和 Key。我用两个文件分工:config.toml管模型和服务的静态配置,settings.json管运行时可能被覆盖的参数和密钥引用。
先看config.toml:
# config.toml —— ELG 推理服务静态配置 [model] # PB 模型路径,导出后的 gaze_opt_b2.m.pb 或 gaze-faceX-lms5.pb pb_path = "./models/gaze_opt_b2.m.pb" # 眼睛输入尺寸,与导出模型一致,顺序为 (height, width) eye_image_shape = [108, 180] # batch 固定为 2,左右眼各一 batch_size = 2 # 输入 OP 名称,替换 Placeholder 后通常是 Placeholder:0 input_op = "Placeholder:0" # 输出 OP 名称 landmarks_op = "upscale/mul:0" radius_op = "radius/out/fc/BiasAdd:0" [server] host = "0.0.0.0" port = 8501 # 单次请求超时,秒 timeout = 15 [taotoken] # 统一 API 通道地址,不带 UTM api_base = "https://taotoken.net/api" # Key 不写在这里,从环境变量或 settings.json 读取 key_env = "TAOTOKEN_API_KEY"再看settings.json,它负责运行时覆盖和密钥引用:
{ "service": { "name": "elg-gaze-infer", "version": "1.0.0", "log_level": "INFO" }, "taotoken": { "api_base": "https://taotoken.net/api", "api_key": "${TAOTOKEN_API_KEY}", "model_channel": "gaze-estimation", "timeout": 15 }, "infer": { "max_batch": 2, "return_heatmap": false, "normalize": true } }${TAOTOKEN_API_KEY}这种写法是给加载器做替换用的,实际读取时用os.environ覆盖。这样做的原因是:配置文件可以进 Git,Key 永远不进。加载逻辑大概长这样:
import json import os import toml def load_config(toml_path="./config.toml", json_path="./settings.json"): with open(toml_path, "r", encoding="utf-8") as f: cfg = toml.load(f) with open(json_path, "r", encoding="utf-8") as f: settings = json.load(f) # 环境变量覆盖 Key key_env = cfg["taotoken"]["key_env"] api_key = os.environ.get(key_env, "") if not api_key: raise RuntimeError(f"环境变量 {key_env} 未设置") settings["taotoken"]["api_key"] = api_key # 合并:settings 优先,config 兜底 cfg["taotoken"]["api_base"] = settings["taotoken"]["api_base"] cfg["taotoken"]["api_key"] = api_key return cfg, settings if __name__ == "__main__": cfg, settings = load_config() print("model pb:", cfg["model"]["pb_path"]) print("api base:", cfg["taotoken"]["api_base"])跑一下这个脚本,能打印出模型路径和 API 地址就说明配置链路通了。这一步别跳过,后面所有报错排查都依赖这两个文件是否被正确加载。
4. 接入推理服务:把 ELG 会话包成受保护接口
配置就绪后,把 ELG 的推理逻辑包成一个 Flask 接口。关键点是:tf.Session()只初始化一次,模型图只导入一次,请求进来只做sess.run。下面是一个最小可用的服务端骨架。
import base64 import numpy as np import cv2 import tensorflow as tf from flask import Flask, request, jsonify from load_config import load_config cfg, settings = load_config() app = Flask(__name__) # 全局会话与 OP SESS = None EYE_OP = None LANDMARKS_OP = None RADIUS_OP = None def init_model(): global SESS, EYE_OP, LANDMARKS_OP, RADIUS_OP SESS = tf.Session() with tf.gfile.FastGFile(cfg["model"]["pb_path"], "rb") as f: graph_def = tf.GraphDef() graph_def.ParseFromString(f.read()) tf.import_graph_def(graph_def, name="") SESS.run(tf.global_variables_initializer()) g = SESS.graph EYE_OP = g.get_tensor_by_name(cfg["model"]["input_op"]) LANDMARKS_OP = g.get_tensor_by_name(cfg["model"]["landmarks_op"]) RADIUS_OP = g.get_tensor_by_name(cfg["model"]["radius_op"]) print("[Info] ELG 模型加载完成") def preprocess_eye(eye_gray): """与训练时一致:直方图均衡 + 归一化到 -1~1""" eye = cv2.equalizeHist(eye_gray) eye = eye.astype(np.float32) eye *= 2.0 / 255.0 eye -= 1.0 return np.expand_dims(eye, -1) @app.route("/v1/gaze/infer", methods=["POST"]) def infer(): # 鉴权:客户端带 TaoToken Key auth = request.headers.get("Authorization", "") if not auth.startswith("Bearer "): return jsonify({"error": "missing bearer token"}), 401 token = auth.split(" ", 1)[1] if token != cfg["taotoken"]["api_key"]: return jsonify({"error": "invalid token"}), 403 body = request.get_json(force=True) # 期望:left_eye / right_eye 为 base64 灰度图 try: left = base64.b64decode(body["left_eye"]) right = base64.b64decode(body["right_eye"]) except Exception: return jsonify({"error": "bad image payload"}), 400 left_img = cv2.imdecode(np.frombuffer(left, np.uint8), cv2.IMREAD_GRAYSCALE) right_img = cv2.imdecode(np.frombuffer(right, np.uint8), cv2.IMREAD_GRAYSCALE) if left_img is None or right_img is None: return jsonify({"error": "decode failed"}), 400 h, w = cfg["model"]["eye_image_shape"] left_img = cv2.resize(left_img, (w, h)) right_img = cv2.resize(right_img, (w, h)) eye1 = preprocess_eye(left_img) eye2 = preprocess_eye(right_img) eyes_batch = np.concatenate((eye1, eye2), axis=0).reshape(2, h, w, 1) landmarks, radius = SESS.run( (LANDMARKS_OP, RADIUS_OP), feed_dict={EYE_OP: eyes_batch} ) return jsonify({ "landmarks": landmarks.tolist(), "radius": radius.tolist(), "shape": list(landmarks.shape) }) if __name__ == "__main__": init_model() app.run(host=cfg["server"]["host"], port=cfg["server"]["port"])这里有几个工程细节值得说。第一,EYE_OP用的是替换后的Placeholder:0,如果你还在用Video/fifo_queue_DequeueMany:1,说明 PB 没做输入 OP 替换,需要先跑一遍pbtxt_to_pb.py那套流程。第二,鉴权只做了一次字符串比对,生产环境建议换成常量时间比较或走网关。第三,landmarks的 shape 是(2, 18, 2),第 0 维是左右眼,第 1 维是 18 个点,第 2 维是 x/y 坐标,这个顺序在返回字段核对时要用到。
启动服务:
export TAOTOKEN_API_KEY="你的Key" python elg_server.py看到ELG 模型加载完成和 Flask 的监听日志,就说明服务起来了。
5. 验证请求与返回字段核对
服务起来后,别急着接前端,先用requests打一次真实请求,把返回字段逐个核对。准备一张包含人脸的图,用 OpenCV 的 LBP 检测器抠出眼睛区域,或者直接用你已经跑通的eyes_detector.py里的crop_eyes输出两张108x180的灰度图。
import base64 import cv2 import requests API = "http://127.0.0.1:8501/v1/gaze/infer" KEY = "你的Key" def to_b64(path): img = cv2.imread(path, cv2.IMREAD_GRAYSCALE) img = cv2.resize(img, (180, 108)) ok, buf = cv2.imencode(".png", img) assert ok return base64.b64encode(buf.tobytes()).decode() payload = { "left_eye": to_b64("left_eye.png"), "right_eye": to_b64("right_eye.png") } resp = requests.post( API, json=payload, headers={"Authorization": f"Bearer {KEY}"}, timeout=15 ) print("status:", resp.status_code) data = resp.json() print("shape:", data["shape"]) print("landmarks[0][16]:", data["landmarks"][0][16]) # 左眼虹膜中心 print("landmarks[1][16]:", data["landmarks"][1][16]) # 右眼虹膜中心 print("radius:", data["radius"])预期结果:status是 200,shape是[2, 18, 2],landmarks[0][16]和landmarks[1][16]是两个长度 2 的数组,代表左右眼虹膜中心在眼睛图像坐标系里的 x/y。radius是[[r_left], [r_right]],单位是像素。
字段核对清单:
| 字段 | 含义 | 预期类型/形状 | 常见异常 |
|---|---|---|---|
| landmarks | 18 个眼区关键点 | list,2x18x2 | 出现 NaN 说明输入归一化不对 |
| landmarks[0:8] | 眼睑 8 点 | 2x8x2 | 顺序错会导致画图错位 |
| landmarks[8:16] | 虹膜 8 点 | 2x8x2 | 与眼睑混淆会画出乱线 |
| landmarks[16] | 虹膜中心 | 2x2 | 这是视线方向的核心 |
| landmarks[17] | 眼睛中心 | 2x2 | 用于计算偏移 |
| radius | 眼睛半径 | 2x1 | 为 0 说明模型没加载对 |
如果landmarks里出现nan,九成是preprocess_eye里equalizeHist之后没做float32转换,或者归一化区间写成了0~1而不是-1~1。如果radius全是 0,检查radius_op的名字是不是radius/out/fc/BiasAdd:0,有些导出流程会把它改名。
6. 本篇常见错排查
报错一:KeyError: 'Placeholder:0'
说明 PB 模型里没有替换后的输入 OP。回到导出流程,确认pbtxt_to_pb.py已经执行,并且get_ops_m里读的是Placeholder:0。如果模型还是Video/fifo_queue_DequeueMany:1,要么重新导出,要么在服务端把input_op改成队列 OP,但后者需要额外喂frame_index和eye_index,不推荐。
报错二:InvalidArgumentError: You must feed a value for placeholder tensor 'Placeholder' with dtype float and shape [2,108,180,1]
输入 shape 对不上。检查eyes_batch.reshape(2, h, w, 1)里的h, w是否和config.toml的eye_image_shape一致。注意顺序是(height, width),即108, 180,别写成180, 108,否则会报 shape 不匹配或者结果全乱。
报错三:请求返回 403invalid token
Key 没对上。先确认export TAOTOKEN_API_KEY和请求头里的 Key 是同一个,再确认服务端加载配置时环境变量确实被读到了。可以在load_config里加一行print(api_key[:6] + "***")做脱敏打印,但别把完整 Key 打进日志。
报错四:cv2.imdecode返回 None
base64 解码后的字节不是合法图片。检查客户端cv2.imencode的格式,PNG 和 JPG 都行,但 base64 字符串不能带data:image/png;base64,前缀,要纯 base64。如果前端传的是带前缀的,服务端要先split(",")。
报错五:并发请求下显存/内存涨得很快
tf.Session()是全局单例,但每次sess.run都会分配临时张量。如果 QPS 高,建议在config.toml里加gpu_options.allow_growth = true,并在服务启动时设置:
gpu_options = tf.GPUOptions(allow_growth=True) SESS = tf.Session(config=tf.ConfigProto(gpu_options=gpu_options))CPU 环境下则关注landmarks.tolist()的序列化开销,必要时改成返回二进制或只返回虹膜中心两个点。
报错六:TaoToken 通道调用超时
如果你是通过 TaoToken 的 API 通道去转发到推理服务,超时时间要设得比模型推理时间长。ELG 单次推理在 CPU 上大概几十到几百毫秒,settings.json里的timeout建议 15 秒起步。如果持续超时,先用curl直连本地服务确认模型本身没问题,再排查通道配置。
7. 下一步:把统一 Key 用到更多链路
到这里,ELG 视线估计的推理服务已经能通过一个 Key 被外部调用了。返回的landmarks和radius可以直接喂给下游的视线方向拟合,也可以存下来做注意力热力图。如果你后面还要接对话模型做“看哪里说哪里”的交互,或者接编码类能力做自动化脚本,没必要再申请第二套凭证——同一个 TaoToken Key 就能覆盖。
需要管理更多 Key 或查看额度,去控制台的 API Keys 页面;接入细节和字段说明以接入文档为准。模型对话相关的调试可以在模型对话页面直接试;如果是要长期跑编码或 Agent 任务,Coding Plan 会更合适。先把今天这套config.toml+settings.json+ Flask 骨架跑通,再按需扩展,比一上来就搭大框架稳得多。