Hugging Face：把本地情感模型包装成 Gradio 网页应用

这篇把 A 篇的本地模型包装成一个 Gradio 网页应用，包含两个功能：单条评价打分（显示三类概率和复核提示），以及批量上传 CSV 打分后下载结果。 应用只监听 127.0.0.1，并关闭了 Gradio 的使用统计上报。测试时启动真实服务，用官方 gradio_client 调用两个 API，覆盖正常结果、空输入、缺少列、Excel 导出的 GBK 编码 CSV 等情况。

先看清成品（box 上真实运行）

• /score_one 返回三类概率。“颜色暗一点但可以接受”的结果是 positive 0.474、neutral 0.462，提示“需要人工复核”。
• /score_csv 上传 CSV 后返回一个结果文件，表头是 text,label,score,needs_review。
• 用户输入有误时，返回中文提示，而不是程序报错堆栈。
• pytest 5 项通过，每次测试都启动真实服务。

合规提醒：Hub 上每个模型和数据集都有自己的许可证，下载前先看卡片。本文用的模型 lxyuan/distilbert-base-multilingual-cased-sentiments-student 是 Apache-2.0，但它的卡片写明是在 tyqiangz/multilingual-sentiments 上蒸馏的，而这个数据集的中文部分来自 amazon_reviews_multi。后者的原始条款是“仅限非商业研究”，并且已经被数据提供方下架。所以商用前请自行评估训练数据的风险，本文只做本地学习演示。Token 不要写进代码或仓库。本文不列 Hugging Face 付费方案的价格，以官网为准。

运行说明：全部在 box 上真实运行。没有部署到 Hugging Face Spaces，因为那是对外发布操作；也没有使用 share=True 生成公网链接。需要对外提供服务时，先加上鉴权和访问频率限制。

做完之后你会得到什么？

需要准备哪些环境？

A 篇的环境和模型，另外 pip install gradio（box 上的版本：gradio 6.30.0、gradio_client 2.7.2）。把 A 篇的 models/ 目录复制过来，或者建一个软链接。

文件结构

56-huggingface-C-src/
├── app.py
├── classify.py
├── test_app.py
└── models -> ../56-huggingface-A-src/models   # 自己建软链接，未随源码附带

第 1 步：应用

"""本地 Gradio 应用：单条文本打分 + 批量 CSV 打分。模型来自 A 篇（models/sentiment），全程离线。"""
import csv, os, tempfile
os.environ.setdefault("GRADIO_ANALYTICS_ENABLED", "False")     # 关闭 Gradio 使用统计上报
import gradio as gr
import classify

CLF = classify.load()
MAX_ROWS = 500

def score_one(text: str):
    text = (text or "").strip()
    if not text:
        raise gr.Error("请输入一段评价")
    (scores,) = CLF([text], truncation=True, max_length=256)
    (r,) = classify.label_rows(CLF, [text])
    return {s["label"]: float(s["score"]) for s in scores}, ("需要人工复核" if r["review"] else "置信度足够")

def read_rows(path):
    """Excel 存的中文 CSV 常是 GBK；依次尝试 utf-8-sig 和 gb18030，都不行再给出明确提示。"""
    raw = open(path, "rb").read()
    for enc in ("utf-8-sig", "gb18030"):
        try:
            return list(csv.DictReader(raw.decode(enc).splitlines()))
        except UnicodeDecodeError:
            continue
    raise gr.Error("无法识别文件编码，请另存为 UTF-8 CSV")

def score_csv(path: str):
    if not path:
        raise gr.Error("请上传 CSV")
    rows = read_rows(path)
    if not rows or "text" not in rows[0]:
        raise gr.Error("CSV 必须有 text 列")
    if len(rows) > MAX_ROWS:
        raise gr.Error(f"一次最多 {MAX_ROWS} 行，当前 {len(rows)} 行")
    res = classify.label_rows(CLF, [r["text"] for r in rows])
    out = tempfile.NamedTemporaryFile("w", suffix=".csv", delete=False, encoding="utf-8", newline="")
    w = csv.writer(out); w.writerow(["text", "label", "score", "needs_review"])
    for r, x in zip(rows, res):
        w.writerow([r["text"], x["label"], x["score"], x["review"]])
    out.close()
    return out.name

with gr.Blocks(title="评价情感打分") as demo:
    gr.Markdown("## 评价情感打分（本地模型，数据不出本机）")
    with gr.Tab("单条"):
        t = gr.Textbox(label="评价内容", lines=3)
        b = gr.Button("打分")
        lab = gr.Label(label="结果", num_top_classes=3)
        note = gr.Textbox(label="提示")
        b.click(score_one, t, [lab, note], api_name="score_one")
    with gr.Tab("批量 CSV"):
        f = gr.File(label="含 text 列的 CSV", file_types=[".csv"], type="filepath")
        b2 = gr.Button("批量打分")
        o = gr.File(label="结果")
        b2.click(score_csv, f, o, api_name="score_csv")

if __name__ == "__main__":
    demo.queue(default_concurrency_limit=2).launch(server_name="127.0.0.1", server_port=int(os.environ.get("APP_PORT", 7861)))

要点：

• api_name 决定了 API 的路径（/score_one、/score_csv），客户端按这个名字调用。
• 用户输入有问题时用 gr.Error 抛出，提示会原样显示给用户。其他异常只返回一句通用提示，不会把内部细节暴露出去（见调试示例）。
• queue(default_concurrency_limit=2) 限制同时推理的请求数，避免 CPU 被占满。

第 2 步：测试

"""启动真实 Gradio 服务，用官方 gradio_client 走一遍 API。"""
import os, pathlib, socket, subprocess, sys, time
import pytest
from gradio_client import Client, handle_file

PORT = 7862

@pytest.fixture(scope="module")
def client():
    env = {**os.environ, "APP_PORT": str(PORT), "GRADIO_ANALYTICS_ENABLED": "False"}
    p = subprocess.Popen([sys.executable, "app.py"], env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    for _ in range(120):
        try:
            socket.create_connection(("127.0.0.1", PORT), 1).close(); break
        except OSError:
            time.sleep(0.5)
    else:
        p.kill(); pytest.fail("服务没有启动")
    yield Client(f"http://127.0.0.1:{PORT}/", verbose=False)
    p.terminate(); p.wait(10)

def test_score_one(client):
    label, note = client.predict("第二次买了，一如既往的好。", api_name="/score_one")
    assert label["label"] == "positive" and note == "置信度足够"

def test_empty_input_is_user_error(client):
    with pytest.raises(Exception, match="请输入一段评价"):
        client.predict("   ", api_name="/score_one")

def test_score_csv(client, tmp_path):
    src = tmp_path / "in.csv"
    src.write_text("text\n用了两天就坏了，非常失望。\n颜色比图片暗一点，不过可以接受。\n", encoding="utf-8")
    out = pathlib.Path(client.predict(handle_file(str(src)), api_name="/score_csv")).read_text(encoding="utf-8").splitlines()
    assert out[0] == "text,label,score,needs_review" and out[1].split(",")[1] == "negative" and len(out) == 3

def test_csv_without_text_column(client, tmp_path):
    src = tmp_path / "bad.csv"; src.write_text("content\nx\n", encoding="utf-8")
    with pytest.raises(Exception, match="text 列"):
        client.predict(handle_file(str(src)), api_name="/score_csv")

def test_gbk_csv_from_excel(client, tmp_path):
    src = tmp_path / "gbk.csv"
    src.write_bytes("text\n用了两天就坏了，非常失望。\n".encode("gbk"))
    out = pathlib.Path(client.predict(handle_file(str(src)), api_name="/score_csv")).read_text(encoding="utf-8")
    assert "negative" in out

第 3 步：运行

$ python app.py            # 浏览器打开 http://127.0.0.1:7861
$ python -c "from gradio_client import Client; print(Client('http://127.0.0.1:7861/').predict('颜色比图片暗一点，不过可以接受。', api_name='/score_one'))"
({'label': 'positive', 'confidences': [{'label': 'positive', 'confidence': 0.47400566935539246}, {'label': 'neutral', 'confidence': 0.461961954832077}, {'label': 'negative', 'confidence': 0.06403238326311111}]}, '需要人工复核')

$ python -m pytest -q
5 passed in 5.08s

调试示例

上传 Excel 导出的 CSV，用户只看到一句含糊的报错。 我用 GBK 编码写了一个 CSV（中文版 Excel 另存为 CSV 时的默认编码）去调用 /score_csv，客户端收到的是：

AppError The upstream Gradio app has raised an exception but has not enabled verbose error reporting. To enable, set show_error=True in launch().

服务端日志里的真正原因是 UnicodeDecodeError: 'utf-8' codec can't decode byte 0xd3 in position 5。我没有按提示开启 show_error=True，因为那会把所有内部异常都暴露给用户。我的做法是新增 read_rows()：先尝试 utf-8-sig，失败再尝试 gb18030，两种都不行就抛出 gr.Error("无法识别文件编码，请另存为 UTF-8 CSV")。同时新增了 test_gbk_csv_from_excel 测试，现在 5 项全部通过。

常见问题

模板规则

• 应用默认只监听本机，并关闭使用统计上报。
• 给用户看的提示用 gr.Error，内部异常只写进日志。
• 用 gradio_client 写端到端测试，每种错误输入都对应一个测试。

项目资源

本地模型加上一个 Gradio 页面，不熟悉代码的同事也能直接上传文件批量打分。