diff --git a/skills/model-labeling/README.md b/skills/model-labeling/README.md
new file mode 100644
index 0000000..8a2337b
--- /dev/null
+++ b/skills/model-labeling/README.md
@@ -0,0 +1,35 @@
+# model-labeling
+
+对已有数据调用线上模型接口批量打标。
+
+典型输入：
+
+- `output/records.jsonl`
+- `output/training.jsonl`
+- `output/eval_planning.csv`
+- `output/records.csv`
+
+典型流程：
+
+1. 确认输入文件路径。
+2. 确认线上模型 `generate` URL。
+3. 用 `batch_label_model.py` 的 `dry_run=true` 识别格式。
+4. 批量请求模型，输出 `output/model_predictions.jsonl`。
+
+执行示例：
+
+```json
+{
+  "script_path": "skills/model-labeling/scripts/batch_label_model.py",
+  "stdin": {
+    "input_path": "output/records.jsonl",
+    "model_url": "http://example/generate",
+    "output_path": "output/model_predictions.jsonl",
+    "parameters": {
+      "max_new_tokens": 64
+    }
+  },
+  "timeout_seconds": 600,
+  "max_output_chars": 20000
+}
+```
diff --git a/skills/model-labeling/SKILL.md b/skills/model-labeling/SKILL.md
new file mode 100644
index 0000000..5ec5d67
--- /dev/null
+++ b/skills/model-labeling/SKILL.md
@@ -0,0 +1,159 @@
+---
+name: model-labeling
+description: 对已有数据集调用线上模型 generate 接口批量打标，支持 canonical records、训练 jsonl、评测 CSV 和同事流转表格的统一归一化。
+when_to_use: 当用户已有一份数据，或刚通过 product-data / online-mining-v2 生成数据后，希望指定线上模型 URL 批量请求模型、得到预测标签、对比真实标签或产出标注结果时使用。
+aliases: batch-labeling, model-annotation, online-model-labeling, 模型打标
+allowed_tools: read_file, write_file, grep_search, glob_search, ask_user_question, python_exec
+---
+
+# Model Labeling
+
+使用这个 skill 处理“已有数据 -> 统一样本格式 -> 调线上模型接口批量打标 -> 输出预测结果”的流程。
+
+线上模型 URL 经常变化，**不要猜 URL**。如果用户没有明确提供 `http://.../generate` 或等价接口地址，必须先询问用户。
+
+## 能力组织
+
+```text
+skills/model-labeling/
+  SKILL.md
+  knowledge/
+    input_formats.md
+  scripts/
+    batch_label_model.py
+```
+
+`batch_label_model.py` 是 portable script，只依赖 Python 标准库。它会：
+
+- 识别 canonical records JSON/JSONL。
+- 识别 product-data 导出的训练 JSONL。
+- 识别 eval/planning CSV，优先使用 `newPrompt`。
+- 识别同事流转 CSV，使用 `query`、`prev_session`、`context`、`function`。
+- 对未知格式返回结构化错误，要求用户提供字段映射，不要擅自转换。
+- 调用用户提供的模型接口，默认请求体为：
+
+```json
+{
+  "inputs": "<prompt>",
+  "parameters": {
+    "max_new_tokens": 64
+  }
+}
+```
+
+## 交互规则
+
+开始执行前必须确认：
+
+1. **输入数据路径**：用户给出的文件路径，或上一轮产物路径。
+2. **模型接口 URL**：必须是用户明确提供的 URL；没有就问。
+3. **输入格式是否可识别**：canonical records、训练 jsonl、eval CSV、同事流转表格可以直接处理。
+4. **未知格式的字段映射**：如果脚本提示 unknown format，需要问用户：
+   - 哪列是 query？
+   - 哪列是 prompt？
+   - 哪列是真实标签？
+   - 是否有历史上下文、context？
+5. **输出路径**：默认写到当前 session 的 `output/model_predictions.jsonl`。
+
+不要在没有 URL 的情况下开始打标。不要把临时结果写到项目根目录。所有输出优先放当前 session 的 `output/`。
+
+## 推荐流程
+
+### 1. 找到输入文件
+
+如果用户说“刚才生成的数据”“这份数据”，先用会话文件面板或 `glob_search` / `read_file` 找到实际路径。常见路径：
+
+```text
+output/records.jsonl
+output/training.jsonl
+output/eval_planning.csv
+output/records.csv
+```
+
+### 2. 确认模型 URL
+
+如果用户没有提供 URL，直接问：
+
+```text
+请提供这次要调用的线上模型 generate 接口 URL，例如 http://.../generate。
+```
+
+如果 `ask_user_question` 可用，优先使用；不可用就普通回复提问并停止。
+
+### 3. 先 dry run 识别格式
+
+先执行一次 `dry_run=true`，只识别格式和样例，不请求模型：
+
+```json
+{
+  "script_path": "skills/model-labeling/scripts/batch_label_model.py",
+  "stdin": {
+    "input_path": "output/records.jsonl",
+    "model_url": "http://example/generate",
+    "dry_run": true,
+    "max_records": 3
+  },
+  "timeout_seconds": 60,
+  "max_output_chars": 20000
+}
+```
+
+如果返回 `ok=false` 且 `needs_mapping=true`，必须把错误和已识别字段展示给用户，让用户说明映射。
+
+### 4. 批量请求模型
+
+确认格式后执行：
+
+```json
+{
+  "script_path": "skills/model-labeling/scripts/batch_label_model.py",
+  "stdin": {
+    "input_path": "output/records.jsonl",
+    "model_url": "http://example/generate",
+    "output_path": "output/model_predictions.jsonl",
+    "parameters": {
+      "max_new_tokens": 64
+    },
+    "timeout_seconds": 60
+  },
+  "timeout_seconds": 600,
+  "max_output_chars": 20000
+}
+```
+
+`output_path` 使用相对 `output/...`，平台会路由到当前 session output 目录。
+
+### 5. 展示结果
+
+执行完成后，简短展示：
+
+- 输入格式。
+- 处理条数、成功数、失败数。
+- 输出路径。
+- 抽 3 条预测样例。
+
+如果存在真实标签，说明输出里包含 `gold_label`，后续可以继续做准确率或错误分析。
+
+## 输出格式
+
+默认输出 JSONL，一行一条紧凑 JSON：
+
+```json
+{"index":0,"request_id":"...","query":"...","gold_label":"complex=false\nAgent(tag=\"地图导航\")","prediction":"...","ok":true,"latency_ms":123}
+```
+
+如果请求失败：
+
+```json
+{"index":0,"query":"...","gold_label":"...","prediction":"","ok":false,"error":"HTTP 500 ...","latency_ms":123}
+```
+
+默认不把完整 prompt 写入结果，避免文件过大。如确实需要排查，可传 `include_prompt=true`。
+
+## 注意事项
+
+- URL、鉴权 header、特殊请求体字段都以用户提供为准。
+- 默认接口字段是 `inputs` 和 `parameters`；如果用户说明接口不同，需要在脚本输入里传 `request_template`。
+- 大批量请求前先小样本 dry run。
+- 打标脚本只负责请求模型和记录预测结果，不负责修改原始数据。
+- 后续准确率统计、错误聚类、补数计划可以再交给其他 skill。
diff --git a/skills/model-labeling/knowledge/input_formats.md b/skills/model-labeling/knowledge/input_formats.md
new file mode 100644
index 0000000..6e702bd
--- /dev/null
+++ b/skills/model-labeling/knowledge/input_formats.md
@@ -0,0 +1,81 @@
+# Model Labeling 输入格式
+
+本 skill 的脚本会把不同来源的数据统一成 sample：
+
+```json
+{
+  "index": 0,
+  "request_id": "",
+  "query": "",
+  "prompt": "",
+  "gold_label": "",
+  "source_format": ""
+}
+```
+
+## canonical records
+
+识别条件：
+
+- JSONL 每行是对象，或 JSON 数组 / `{ "records": [...] }`。
+- 对象包含 `turn.query`。
+
+字段来源：
+
+- `query`：`turn.query`
+- `request_id`：`source.request_id`
+- `gold_label`：`complex=true/false` + `label.target`
+- `prompt`：按 product-data 的 planning prompt 规则生成
+
+## training jsonl
+
+识别条件：
+
+- 每行是对象。
+- 包含 `instruction` 或 `system`。
+
+字段来源：
+
+- `prompt`：`system + instruction` 包成 chat template；如果已有 `prompt` 则直接使用。
+- `gold_label`：`output`
+- `query`：尽力从 `[当前query]` 后的 `用户:` 抽取。
+
+## eval/planning CSV
+
+识别条件：
+
+- CSV 表头包含 `newPrompt` 或 `query`。
+
+字段来源：
+
+- `prompt`：优先 `newPrompt`
+- `query`：`query`
+- `gold_label`：`code标签`，如果有 `complex` 列则组合成两行
+
+## 同事流转 CSV
+
+识别条件：
+
+- CSV 表头包含 `query` 和 `function`。
+
+字段来源：
+
+- `query`：`query`
+- `request_id`：`request_id`
+- `gold_label`：`function`
+- `prompt`：根据 `query`、`prev_session`、`context` 生成 planning prompt
+
+## 未知格式
+
+如果不能识别，脚本返回：
+
+```json
+{
+  "ok": false,
+  "needs_mapping": true,
+  "columns": ["..."],
+  "error": "..."
+}
+```
+
+此时必须询问用户字段映射，不要猜。
diff --git a/skills/model-labeling/scripts/batch_label_model.py b/skills/model-labeling/scripts/batch_label_model.py
new file mode 100755
index 0000000..f177fab
--- /dev/null
+++ b/skills/model-labeling/scripts/batch_label_model.py
@@ -0,0 +1,600 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+"""批量调用线上模型 generate 接口给数据打标。
+
+输入通过 stdin 或 --input 传 JSON 对象，输出稳定 JSON 对象到 stdout。
+脚本只依赖 Python 标准库，便于在本地、Linux runtime 用户或远程工作区迁移执行。
+"""
+
+import argparse
+import csv
+import json
+import os
+import re
+import sys
+import time
+import urllib.error
+import urllib.request
+from io import StringIO
+from pathlib import Path
+from typing import Any
+
+
+DEFAULT_SYSTEM_PROMPT = "你是小爱同学，中文智能语音助手。"
+DEFAULT_PARAMETERS = {"max_new_tokens": 64}
+
+
+class LabelingError(ValueError):
+    """输入参数或数据格式错误。"""
+
+
+def main() -> int:
+    try:
+        payload = load_payload()
+        result = run(payload)
+        emit({"ok": True, **result})
+        return 0
+    except LabelingError as exc:
+        emit({"ok": False, "error": str(exc), **getattr(exc, "extra", {})})
+        return 1
+    except Exception as exc:  # noqa: BLE001 - CLI 需要稳定 JSON 错误
+        emit({"ok": False, "error": str(exc)})
+        return 1
+
+
+def load_payload() -> dict[str, Any]:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--input", "-i", help="JSON 参数文件；不传则从 stdin 读取")
+    args = parser.parse_args()
+    text = Path(args.input).read_text(encoding="utf-8") if args.input else sys.stdin.read()
+    if not text.strip():
+        raise LabelingError("input JSON is required")
+    payload = json.loads(text)
+    if not isinstance(payload, dict):
+        raise LabelingError("input JSON must be an object")
+    return payload
+
+
+def emit(payload: dict[str, Any]) -> None:
+    print(json.dumps(payload, ensure_ascii=False, separators=(",", ":")))
+
+
+def run(payload: dict[str, Any]) -> dict[str, Any]:
+    input_path = require_string(payload, "input_path")
+    model_url = require_string(payload, "model_url")
+    dry_run = bool(payload.get("dry_run", False))
+    include_prompt = bool(payload.get("include_prompt", False))
+    max_records = optional_int(payload.get("max_records"))
+    start_index = int(payload.get("start_index") or 0)
+    timeout_seconds = float(payload.get("timeout_seconds") or 60)
+    output_path = str(payload.get("output_path") or "output/model_predictions.jsonl")
+    parameters = payload.get("parameters")
+    if parameters is None:
+        parameters = dict(DEFAULT_PARAMETERS)
+    if not isinstance(parameters, dict):
+        raise LabelingError("parameters must be an object")
+    headers = payload.get("headers")
+    if headers is None:
+        headers = {"Content-Type": "application/json"}
+    if not isinstance(headers, dict):
+        raise LabelingError("headers must be an object")
+    request_template = payload.get("request_template")
+    if request_template is not None and not isinstance(request_template, dict):
+        raise LabelingError("request_template must be an object")
+
+    source = read_input_file(input_path)
+    samples = normalize_samples(
+        source,
+        field_mapping=payload.get("field_mapping"),
+        system_prompt=str(payload.get("system_prompt") or DEFAULT_SYSTEM_PROMPT),
+        session_num=int(payload.get("session_num") or 5),
+        session_time_minutes=int(payload.get("session_time_minutes") or 5),
+    )
+    if start_index:
+        samples = samples[start_index:]
+    if max_records is not None:
+        samples = samples[:max_records]
+
+    summary = {
+        "input_path": input_path,
+        "source_format": source["format"],
+        "total_samples": len(samples),
+        "sample_preview": preview_samples(samples),
+    }
+    if dry_run:
+        return {**summary, "dry_run": True}
+
+    if not model_url.startswith(("http://", "https://")):
+        raise LabelingError("model_url must start with http:// or https://")
+    results: list[dict[str, Any]] = []
+    ok_count = 0
+    output = resolve_runtime_path(output_path)
+    output.parent.mkdir(parents=True, exist_ok=True)
+    with output.open("w", encoding="utf-8") as file:
+        for sample in samples:
+            item = request_one(
+                sample,
+                model_url=model_url,
+                parameters=parameters,
+                headers={str(k): str(v) for k, v in headers.items()},
+                timeout_seconds=timeout_seconds,
+                request_template=request_template,
+                include_prompt=include_prompt,
+            )
+            if item.get("ok"):
+                ok_count += 1
+            results.append(item)
+            file.write(json.dumps(item, ensure_ascii=False, separators=(",", ":")) + "\n")
+
+    return {
+        **summary,
+        "dry_run": False,
+        "output_path": str(output),
+        "success_count": ok_count,
+        "failure_count": len(results) - ok_count,
+        "result_preview": results[:3],
+    }
+
+
+def request_one(
+    sample: dict[str, Any],
+    *,
+    model_url: str,
+    parameters: dict[str, Any],
+    headers: dict[str, str],
+    timeout_seconds: float,
+    request_template: dict[str, Any] | None,
+    include_prompt: bool,
+) -> dict[str, Any]:
+    prompt = str(sample["prompt"])
+    body = build_request_body(prompt, parameters, request_template)
+    started = time.time()
+    base = {
+        "index": sample["index"],
+        "request_id": sample.get("request_id", ""),
+        "query": sample.get("query", ""),
+        "gold_label": sample.get("gold_label", ""),
+        "source_format": sample.get("source_format", ""),
+    }
+    if include_prompt:
+        base["prompt"] = prompt
+    try:
+        request = urllib.request.Request(
+            model_url,
+            data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
+            headers=headers,
+            method="POST",
+        )
+        with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
+            text = response.read().decode("utf-8", errors="replace")
+            status = getattr(response, "status", 200)
+        latency_ms = int((time.time() - started) * 1000)
+        decoded = try_json(text)
+        return {
+            **base,
+            "prediction": extract_prediction(decoded, text),
+            "ok": 200 <= int(status) < 300,
+            "status": int(status),
+            "latency_ms": latency_ms,
+            "response": decoded if decoded is not None else text,
+        }
+    except urllib.error.HTTPError as exc:
+        text = exc.read().decode("utf-8", errors="replace")
+        return {
+            **base,
+            "prediction": extract_prediction(try_json(text), text),
+            "ok": False,
+            "status": exc.code,
+            "latency_ms": int((time.time() - started) * 1000),
+            "error": f"HTTP {exc.code}: {text[:500]}",
+        }
+    except Exception as exc:  # noqa: BLE001 - 单条失败不中断整体批次
+        return {
+            **base,
+            "prediction": "",
+            "ok": False,
+            "latency_ms": int((time.time() - started) * 1000),
+            "error": str(exc),
+        }
+
+
+def build_request_body(
+    prompt: str,
+    parameters: dict[str, Any],
+    request_template: dict[str, Any] | None,
+) -> dict[str, Any]:
+    if request_template is None:
+        return {"inputs": prompt, "parameters": parameters}
+    return replace_placeholders(request_template, {"prompt": prompt, "parameters": parameters})
+
+
+def replace_placeholders(value: Any, variables: dict[str, Any]) -> Any:
+    if isinstance(value, str):
+        if value == "{{prompt}}":
+            return variables["prompt"]
+        if value == "{{parameters}}":
+            return variables["parameters"]
+        return value.replace("{{prompt}}", str(variables["prompt"]))
+    if isinstance(value, list):
+        return [replace_placeholders(item, variables) for item in value]
+    if isinstance(value, dict):
+        return {key: replace_placeholders(item, variables) for key, item in value.items()}
+    return value
+
+
+def read_input_file(path: str) -> dict[str, Any]:
+    file_path = resolve_runtime_path(path)
+    if not file_path.exists():
+        raise LabelingError(f"input_path not found: {path}")
+    suffix = file_path.suffix.lower()
+    text = file_path.read_text(encoding="utf-8", errors="replace")
+    if suffix in {".csv", ".tsv"}:
+        delimiter = "\t" if suffix == ".tsv" else ","
+        rows = list(csv.DictReader(StringIO(text), delimiter=delimiter))
+        return {"format": "table", "path": str(file_path), "rows": rows, "columns": list(rows[0].keys()) if rows else []}
+    if suffix == ".jsonl":
+        rows = [json.loads(line) for line in text.splitlines() if line.strip()]
+        return {"format": "jsonl", "path": str(file_path), "rows": rows}
+    decoded = json.loads(text)
+    if isinstance(decoded, dict) and isinstance(decoded.get("records"), list):
+        decoded = decoded["records"]
+    if isinstance(decoded, list):
+        return {"format": "json", "path": str(file_path), "rows": decoded}
+    raise_with_mapping("JSON input must be an array or {records:[...]}", columns=list(decoded.keys()) if isinstance(decoded, dict) else [])
+
+
+def normalize_samples(
+    source: dict[str, Any],
+    *,
+    field_mapping: Any,
+    system_prompt: str,
+    session_num: int,
+    session_time_minutes: int,
+) -> list[dict[str, Any]]:
+    rows = source.get("rows")
+    if not isinstance(rows, list) or not rows:
+        raise LabelingError("input file contains no rows")
+    if isinstance(field_mapping, dict):
+        return samples_from_mapping(rows, field_mapping, source["format"], system_prompt)
+    first = rows[0]
+    if not isinstance(first, dict):
+        raise_with_mapping("rows must be objects")
+    if is_canonical_record(first):
+        return [
+            sample_from_record(index, row, system_prompt, session_num, session_time_minutes)
+            for index, row in enumerate(rows)
+            if isinstance(row, dict)
+        ]
+    if is_training_row(first):
+        return [sample_from_training(index, row, system_prompt) for index, row in enumerate(rows) if isinstance(row, dict)]
+    if source["format"] == "table":
+        columns = list(first.keys())
+        if "newPrompt" in columns:
+            return [sample_from_eval_row(index, row) for index, row in enumerate(rows) if isinstance(row, dict)]
+        if "query" in columns and "function" in columns:
+            return [sample_from_flow_row(index, row, system_prompt) for index, row in enumerate(rows) if isinstance(row, dict)]
+        raise_with_mapping("unrecognized table format", columns=columns)
+    raise_with_mapping("unrecognized JSON/JSONL format", columns=list(first.keys()))
+
+
+def sample_from_record(
+    index: int,
+    record: dict[str, Any],
+    system_prompt: str,
+    session_num: int,
+    session_time_minutes: int,
+) -> dict[str, Any]:
+    source = record.get("source") if isinstance(record.get("source"), dict) else {}
+    turn = record.get("turn") if isinstance(record.get("turn"), dict) else {}
+    return {
+        "index": index,
+        "request_id": str(source.get("request_id") or ""),
+        "query": str(turn.get("query") or ""),
+        "prompt": build_planning_prompt(record, system_prompt, session_num, session_time_minutes),
+        "gold_label": combined_label(record),
+        "source_format": "canonical_record_v1",
+    }
+
+
+def sample_from_training(index: int, row: dict[str, Any], system_prompt: str) -> dict[str, Any]:
+    prompt = str(row.get("prompt") or "")
+    if not prompt:
+        system = str(row.get("system") or system_prompt)
+        instruction = str(row.get("instruction") or row.get("input") or "")
+        prompt = wrap_chat_prompt(system, instruction)
+    return {
+        "index": index,
+        "request_id": str(row.get("request_id") or ""),
+        "query": extract_query_from_prompt(prompt),
+        "prompt": prompt,
+        "gold_label": str(row.get("output") or row.get("target") or ""),
+        "source_format": "training_jsonl",
+    }
+
+
+def sample_from_eval_row(index: int, row: dict[str, Any]) -> dict[str, Any]:
+    gold = str(row.get("code标签") or row.get("function") or row.get("target") or "")
+    complex_value = row.get("complex")
+    if complex_value not in (None, "") and not gold.startswith("complex="):
+        gold = f"complex={normalize_bool_literal(complex_value)}\n{gold}".rstrip()
+    prompt = str(row.get("newPrompt") or row.get("prompt") or "")
+    query = str(row.get("query") or "")
+    if not prompt:
+        prompt = build_planning_prompt(minimal_record(query, {}, [], gold), DEFAULT_SYSTEM_PROMPT, 5, 5)
+    return {
+        "index": index,
+        "request_id": str(row.get("request_id") or ""),
+        "query": query,
+        "prompt": prompt,
+        "gold_label": gold,
+        "source_format": "eval_csv",
+    }
+
+
+def sample_from_flow_row(index: int, row: dict[str, Any], system_prompt: str) -> dict[str, Any]:
+    record = record_from_flow_row(row)
+    return {
+        "index": index,
+        "request_id": str(row.get("request_id") or ""),
+        "query": str(row.get("query") or ""),
+        "prompt": build_planning_prompt(record, system_prompt, 5, 5),
+        "gold_label": str(row.get("function") or ""),
+        "source_format": "flow_csv",
+    }
+
+
+def samples_from_mapping(
+    rows: list[Any],
+    mapping: dict[str, Any],
+    source_format: str,
+    system_prompt: str,
+) -> list[dict[str, Any]]:
+    query_field = str(mapping.get("query") or "")
+    prompt_field = str(mapping.get("prompt") or "")
+    label_field = str(mapping.get("label") or mapping.get("target") or "")
+    request_id_field = str(mapping.get("request_id") or "")
+    if not query_field and not prompt_field:
+        raise LabelingError("field_mapping must provide query or prompt")
+    samples: list[dict[str, Any]] = []
+    for index, row in enumerate(rows):
+        if not isinstance(row, dict):
+            continue
+        prompt = str(row.get(prompt_field) or "")
+        query = str(row.get(query_field) or "")
+        if not prompt:
+            record = minimal_record(query, {}, [], str(row.get(label_field) or ""))
+            prompt = build_planning_prompt(record, system_prompt, 5, 5)
+        samples.append(
+            {
+                "index": index,
+                "request_id": str(row.get(request_id_field) or ""),
+                "query": query or extract_query_from_prompt(prompt),
+                "prompt": prompt,
+                "gold_label": str(row.get(label_field) or ""),
+                "source_format": f"{source_format}_mapped",
+            }
+        )
+    return samples
+
+
+def is_canonical_record(row: dict[str, Any]) -> bool:
+    return isinstance(row.get("turn"), dict) and bool(row["turn"].get("query"))
+
+
+def is_training_row(row: dict[str, Any]) -> bool:
+    return any(key in row for key in ("instruction", "system", "output", "prompt"))
+
+
+def build_planning_prompt(
+    record: dict[str, Any],
+    system_prompt: str,
+    session_num: int,
+    session_time_minutes: int,
+) -> str:
+    instruction = build_training_instruction(record, session_num, session_time_minutes)
+    return wrap_chat_prompt(system_prompt, instruction)
+
+
+def wrap_chat_prompt(system_prompt: str, instruction: str) -> str:
+    return (
+        f"<|im_start|>system\n{system_prompt}<|im_end|>\n"
+        f"<|im_start|>user\n{instruction}<|im_end|>\n"
+        "<|im_start|>assistant\n"
+    )
+
+
+def build_training_instruction(record: dict[str, Any], session_num: int, session_time_minutes: int) -> str:
+    turn = record.get("turn") if isinstance(record.get("turn"), dict) else {}
+    query = str(turn.get("query") or "")
+    context = record.get("context") if isinstance(record.get("context"), dict) else {}
+    prev_session = record.get("prev_session") if isinstance(record.get("prev_session"), list) else []
+    current_ts = optional_int(turn.get("timestamp"))
+    history = render_history(prev_session, current_ts, session_num, session_time_minutes)
+    return (
+        "请参考用户的[当前query]、[对话历史]、[知识注入]、[系统状态]识别出[当前query]的[function]结果，[function]是python的code形式。\n"
+        "[知识注入]\n"
+        f"{json.dumps({'location': str(context.get('location') or ''), 'rag': str(context.get('rag') or '')}, ensure_ascii=False, indent=0)}\n"
+        "[系统状态]\n"
+        "{}\n"
+        "[对话历史]\n"
+        f"{history}"
+        "[当前query]\n"
+        f"用户: {query}\n"
+        "[function]\n"
+    )
+
+
+def render_history(prev_session: list[Any], current_ts: int | None, session_num: int, session_time_minutes: int) -> str:
+    usable: list[dict[str, Any]] = []
+    for item in prev_session[-session_num:]:
+        if not isinstance(item, dict):
+            continue
+        ts = optional_int(item.get("timestamp"))
+        if current_ts is not None and ts is not None:
+            if abs(current_ts - ts) > session_time_minutes * 60_000:
+                continue
+        usable.append(item)
+    if not usable:
+        return ""
+    lines: list[str] = []
+    for item in usable:
+        query = str(item.get("query") or "").strip()
+        tts = str(item.get("tts") or "").strip()
+        if query:
+            lines.append(f"用户: {query}")
+        if tts:
+            lines.append(f"小爱: {tts}")
+    return "\n".join(lines) + ("\n" if lines else "")
+
+
+def record_from_flow_row(row: dict[str, Any]) -> dict[str, Any]:
+    prev_session = parse_json_cell(row.get("prev_session"), default=[])
+    context = parse_json_cell(row.get("context"), default={})
+    return minimal_record(
+        str(row.get("query") or ""),
+        context if isinstance(context, dict) else {},
+        prev_session if isinstance(prev_session, list) else [],
+        str(row.get("function") or ""),
+        request_id=str(row.get("request_id") or ""),
+        timestamp=optional_int(row.get("timestamp")),
+    )
+
+
+def minimal_record(
+    query: str,
+    context: dict[str, Any],
+    prev_session: list[Any],
+    target: str,
+    *,
+    request_id: str = "",
+    timestamp: int | None = None,
+) -> dict[str, Any]:
+    return {
+        "source": {"request_id": request_id, "timestamp": timestamp},
+        "turn": {"query": query, "timestamp": timestamp},
+        "prev_session": prev_session,
+        "context": context,
+        "label": {"target": target},
+        "dimensions": {},
+    }
+
+
+def combined_label(record: dict[str, Any]) -> str:
+    target = ""
+    label = record.get("label") if isinstance(record.get("label"), dict) else {}
+    if isinstance(label, dict):
+        target = str(label.get("target") or "")
+    dimensions = record.get("dimensions") if isinstance(record.get("dimensions"), dict) else {}
+    complex_value = dimensions.get("complex") if isinstance(dimensions, dict) else None
+    if isinstance(complex_value, bool):
+        return f"complex={'true' if complex_value else 'false'}\n{target}".rstrip()
+    return target
+
+
+def extract_query_from_prompt(prompt: str) -> str:
+    match = re.search(r"\[当前query\]\s*\n用户[:：]\s*(.+)", prompt)
+    return match.group(1).strip() if match else ""
+
+
+def normalize_bool_literal(value: Any) -> str:
+    text = str(value).strip().lower()
+    return "true" if text in {"true", "1", "yes", "y", "是", "复杂"} else "false"
+
+
+def parse_json_cell(value: Any, default: Any) -> Any:
+    if value in (None, ""):
+        return default
+    if isinstance(value, (dict, list)):
+        return value
+    try:
+        return json.loads(str(value))
+    except json.JSONDecodeError:
+        return default
+
+
+def resolve_runtime_path(path: str) -> Path:
+    raw = Path(path).expanduser()
+    if raw.is_absolute():
+        return raw
+    scratchpad = Path(str(Path.cwd()))
+    if os.environ.get("PYTHON_EXEC_SCRATCHPAD"):
+        scratchpad = Path(os.environ["PYTHON_EXEC_SCRATCHPAD"]).expanduser()
+    parts = raw.parts
+    if parts and parts[0] in {"output", "outputs"}:
+        return scratchpad.parent / "output" / Path(*parts[1:])
+    if parts and parts[0] in {"input", "inputs"}:
+        return scratchpad.parent / "input" / Path(*parts[1:])
+    if parts and parts[0] in {"scratchpad", "scratch"}:
+        return scratchpad / Path(*parts[1:])
+    return raw
+
+
+def optional_int(value: Any) -> int | None:
+    if value in (None, ""):
+        return None
+    try:
+        return int(float(str(value)))
+    except (TypeError, ValueError):
+        return None
+
+
+def extract_prediction(decoded: Any, text: str) -> str:
+    if isinstance(decoded, dict):
+        for key in ("generated_text", "text", "output", "response", "result"):
+            value = decoded.get(key)
+            if isinstance(value, str):
+                return value.strip()
+        outputs = decoded.get("outputs")
+        if isinstance(outputs, list) and outputs:
+            first = outputs[0]
+            if isinstance(first, str):
+                return first.strip()
+            if isinstance(first, dict):
+                return extract_prediction(first, json.dumps(first, ensure_ascii=False))
+        choices = decoded.get("choices")
+        if isinstance(choices, list) and choices:
+            first = choices[0]
+            if isinstance(first, dict):
+                message = first.get("message")
+                if isinstance(message, dict) and isinstance(message.get("content"), str):
+                    return message["content"].strip()
+                if isinstance(first.get("text"), str):
+                    return first["text"].strip()
+    return text.strip()
+
+
+def try_json(text: str) -> Any:
+    try:
+        return json.loads(text)
+    except json.JSONDecodeError:
+        return None
+
+
+def preview_samples(samples: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    return [
+        {
+            "index": item.get("index"),
+            "query": item.get("query"),
+            "gold_label": item.get("gold_label"),
+            "prompt_preview": str(item.get("prompt") or "")[:200],
+        }
+        for item in samples[:3]
+    ]
+
+
+def require_string(payload: dict[str, Any], key: str) -> str:
+    value = payload.get(key)
+    if not isinstance(value, str) or not value.strip():
+        raise LabelingError(f"{key} is required")
+    return value.strip()
+
+
+def raise_with_mapping(message: str, columns: list[str] | None = None) -> None:
+    exc = LabelingError(message)
+    exc.extra = {"needs_mapping": True, "columns": columns or []}  # type: ignore[attr-defined]
+    raise exc
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())