1. Airflow 任务超时没人管？先看清及时性校验要解决什么
Airflow 的 DAG 跑起来之后，最让人头疼的不是任务失败，而是任务「没失败但也没按时完成」。失败会触发 retry、会亮红灯，可一个任务卡在 running 状态、或者上游延迟导致下游整体后移，调度器本身不会主动告诉你。等到业务方来问「今天报表怎么还没出」，你才发现某个 task 已经晚了两个小时。
这就是任务及时性校验要解决的问题：不是判断任务成功或失败，而是判断任务有没有在预期时间点之前完成。它和 Airflow 自带的 SLA 机制有区别——SLA 更偏向「任务超过预期时长就告警」，而及时性校验关注的是「某个业务时间点之前必须完成」，比如每天 01:45 前必须跑完的结算任务、每 10 分钟必须刷新一次的特征表。
适合谁看这篇：已经在用 Airflow 做调度、手上有几十到几百个 DAG、需要一套统一告警出口的运维或数据开发。核心检索词就是 Python、airflow、任务及时性校验，我会用 Python 写校验逻辑，用 TaoToken 统一 Key 打通告警链路，避免每个 DAG 里散落一堆 webhook 和密钥。
传统做法通常是在每个 DAG 里塞一个 PythonOperator，任务跑完就发钉钉。问题是：告警渠道的密钥散落在各个 DAG、换一个告警通道要改几十个文件、校验逻辑重复且难维护。更麻烦的是，很多团队想接大模型做告警内容润色或根因初判，结果每个 DAG 都要配一份 API Key，管理成本直接爆炸。
我的思路是把「校验」和「通知」拆开：校验逻辑集中在一个独立的 DAG 里，定时扫描元数据库和业务表，判断哪些任务该完成却没完成；通知统一走 TaoToken 的 API 通道，一个 Key 覆盖模型调用和告警分发。这样新增一个监控任务只需要往表里插一行配置，不用动代码。
下面会交付四块可复制内容：建表与配置数据、及时性校验的 Python 逻辑、TaoToken 的配置片段、本地触发验证步骤。你照着做就能跑通一条从「发现超时」到「发出告警」的完整链路。
2. 用 TaoToken 统一 Key 打通告警链路的前置准备
在写校验代码之前，先把告警出口统一掉。很多人的 Airflow 告警是「一个 DAG 一个 webhook」，钉钉、飞书、邮件各配各的，密钥明文写在 DAG 文件里。一旦要换通道或者加一个大模型润色环节，改动面非常大。
TaoToken 在这里扮演的角色是统一 API 通道：你只需要一个 Key，就能调用模型对话能力，也能作为告警服务的统一入口。官网在 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 地址是 https://taotoken.net/api 。注意 API 地址不带 UTM 参数，配置时直接用这个。
前置准备分三步。
第一步，拿到 Key。登录后进入控制台，在 API Keys 页面创建一个新 Key。建议按用途命名，比如airflow-alert，方便后续轮换。创建后立刻复制保存，页面刷新后就不再完整显示。控制台地址是 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite 。
第二步，确认模型 ID。及时性校验的告警内容润色、根因初判可以用模型对话能力完成。你可以在模型对话页面先试一下可用模型，地址是 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite 。记下你要用的 Model ID，后面配置里要填。
第三步，决定配置存放位置。不要把 Key 写死在 DAG 文件里。推荐两种方式：一是放在 Airflow 的 Connections 里，类型选 HTTP，Host 填https://taotoken.net/api，Password 填 Key；二是放在环境变量里，通过AIRFLOW_VAR_前缀注入。我实测下来环境变量方式最省事，本地和线上都好切换。
如果你后续要做长期编码或 Agent 类任务，可以了解 Coding Plan，地址是 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 。但本篇的告警链路用按量 API 就够了。
这里要强调一个原则：TaoToken 是 API 通道，不是替代 Airflow 调度器，也不是替代你的告警服务。它解决的是「密钥统一」和「模型能力接入」两件事。校验逻辑、调度触发仍然由 Airflow 负责。
配置片段我会在下一节给出，包含 Base URL、Key、Model ID 三件套，这是接入任何 OpenAI 兼容接口的标准结构。你先把 Key 准备好，我们进入建表和代码环节。
3. 可复制的建表 SQL 与 TaoToken 配置片段
这一节给出可以直接粘贴运行的内容。先建三张表：通知组表、任务清单表、DAG 调度时间表。它们的关系是：任务清单表记录「哪个 DAG 的哪个 task 应该在什么时间前完成」，通过 notify_group_id 关联到通知组，通知组里存告警出口配置。
建表 SQL 如下，注意字符集和注释：
CREATE TABLE IF NOT EXISTS notify_group ( id INT(11) NOT NULL COMMENT '通知组表id', email_addr_list VARCHAR(500) COMMENT '负责人邮箱清单', phone_list VARCHAR(500) COMMENT '负责人电话清单', webhook VARCHAR(500) COMMENT '告警机器人完整链接', secret VARCHAR(500) COMMENT '告警机器人签名密钥', PRIMARY KEY (id) ) DEFAULT CHARSET=utf8mb4 COMMENT='通知组表'; CREATE TABLE IF NOT EXISTS task_list ( id INT(11) NOT NULL COMMENT '任务清单表id', dag_id VARCHAR(50) DEFAULT NULL COMMENT 'DAG id', task_id VARCHAR(500) DEFAULT NULL COMMENT '任务id，多个用逗号分隔', latest_complete_time VARCHAR(10) DEFAULT NULL COMMENT '最晚完成时间 HH:mm', notify_group_id INT(11) DEFAULT NULL COMMENT '通知组表id', PRIMARY KEY (id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='任务清单表'; CREATE TABLE IF NOT EXISTS dag_list ( id INT(11) NOT NULL COMMENT 'id', dag_id VARCHAR(50) DEFAULT NULL COMMENT 'DAG id', schedule_time VARCHAR(10) DEFAULT NULL COMMENT '调度时间 HH:mm', PRIMARY KEY (id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='DAG调度时间表';插入一条测试配置，把 webhook 换成你自己的告警机器人地址：
INSERT INTO notify_group (id, email_addr_list, phone_list, webhook, secret) VALUES (1, 'data-oncall@example.com', '18800000000', 'https://oapi.dingtalk.com/robot/send?access_token=your_token', 'your_secret'); INSERT INTO task_list (id, dag_id, task_id, latest_complete_time, notify_group_id) VALUES (1, 'daily_settlement', 'settle_task,report_task', '01:45', 1); INSERT INTO dag_list (id, dag_id, schedule_time) VALUES (1, 'daily_settlement', '01:00');接下来是 TaoToken 配置片段。推荐用 JSON 存环境变量，Airflow 读取后注入。文件路径放在$AIRFLOW_HOME/config/taotoken.json：
{ "base_url": "https://taotoken.net/api", "api_key": "sk-your-taotoken-key", "model_id": "your-model-id", "timeout": 30, "max_retries": 2 }如果你用 Airflow Connections，配置等价于：
[conn_id: taotoken_api] conn_type = http host = https://taotoken.net/api password = sk-your-taotoken-key extra = {"model_id": "your-model-id", "timeout": 30}三件套对照表，方便你核对：
| 配置项 | 值 | 说明 |
|---|---|---|
| Base URL | https://taotoken.net/api | 不带 UTM，固定 |
| API Key | sk-开头 | 控制台创建，按用途命名 |
| Model ID | 模型对话页获取 | 用于告警润色与根因初判 |
注意：Key 不要提交到 Git。用.gitignore排除配置文件，线上通过环境变量或 Airflow 的 secrets backend 注入。我踩过的坑是把 Key 写进 DAG 默认参数，结果日志里打印出来了，后来改成只从环境变量读。
配置就绪后，下一节写校验逻辑。核心是查 task_list 和 dag_list，结合 Airflow 元数据库里的 task_instance 状态，判断是否超时。
4. 及时性校验 DAG 的完整 Python 代码与本地触发验证
校验逻辑放在一个独立的 DAG 里，每 10 分钟跑一次。它做四件事：读取配置表、查询 Airflow 元数据库里对应 task 的最新状态、判断是否超过 latest_complete_time、超时则调用 TaoToken 发告警。
先写数据库连接和配置读取模块timeliness_check.py：
import json import os from datetime import datetime, timedelta import pymysql from airflow import DAG from airflow.operators.python import PythonOperator from airflow.utils.dates import days_ago CONFIG_PATH = os.path.join(os.environ.get("AIRFLOW_HOME", "/opt/airflow"), "config", "taotoken.json") def load_taotoken_config(): with open(CONFIG_PATH, "r", encoding="utf-8") as f: return json.load(f) def get_biz_conn(): return pymysql.connect( host=os.environ["BIZ_DB_HOST"], user=os.environ["BIZ_DB_USER"], password=os.environ["BIZ_DB_PASSWORD"], database=os.environ["BIZ_DB_NAME"], charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, ) def get_airflow_conn(): return pymysql.connect( host=os.environ["AIRFLOW_DB_HOST"], user=os.environ["AIRFLOW_DB_USER"], password=os.environ["AIRFLOW_DB_PASSWORD"], database=os.environ.get("AIRFLOW_DB_NAME", "airflow"), charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, )注意 pymysql 连接必须用关键字参数，写成pymysql.connect(host=..., user=...)。如果写成位置参数会报__init__() takes 1 positional argument but 5 were given，这是新手最常见的错误。
接着写核心校验函数：
def check_timeliness(**context): now = datetime.now() today = now.strftime("%Y-%m-%d") biz_conn = get_biz_conn() af_conn = get_airflow_conn() overdue_tasks = [] with biz_conn.cursor() as cur: cur.execute(""" SELECT t.id, t.dag_id, t.task_id, t.latest_complete_time, n.webhook, n.secret, n.email_addr_list FROM task_list t JOIN notify_group n ON t.notify_group_id = n.id """) configs = cur.fetchall() for cfg in configs: deadline = datetime.strptime( f"{today} {cfg['latest_complete_time']}", "%Y-%m-%d %H:%M" ) if now < deadline: continue task_ids = [x.strip() for x in cfg["task_id"].split(",") if x.strip()] with af_conn.cursor() as cur: placeholders = ",".join(["%s"] * len(task_ids)) cur.execute(f""" SELECT task_id, state, end_date FROM task_instance WHERE dag_id = %s AND task_id IN ({placeholders}) AND execution_date >= %s ORDER BY execution_date DESC """, [cfg["dag_id"]] + task_ids + [deadline - timedelta(hours=6)]) rows = cur.fetchall() done = {r["task_id"] for r in rows if r["state"] == "success"} missing = [t for t in task_ids if t not in done] if missing: overdue_tasks.append({ "dag_id": cfg["dag_id"], "missing": missing, "deadline": cfg["latest_complete_time"], "webhook": cfg["webhook"], "secret": cfg["secret"], }) biz_conn.close() af_conn.close() if overdue_tasks: send_alert(overdue_tasks) return overdue_tasks告警发送函数走 TaoToken，先让模型把告警内容整理成人话，再发到 webhook：
import requests def send_alert(overdue_tasks): cfg = load_taotoken_config() summary = "\n".join( f"DAG {t['dag_id']} 未在 {t['deadline']} 前完成，缺失任务：{','.join(t['missing'])}" for t in overdue_tasks ) prompt = f"请把以下 Airflow 任务超时信息整理成一条简洁的中文告警，包含影响和建议动作：\n{summary}" resp = requests.post( f"{cfg['base_url']}/v1/chat/completions", headers={ "Authorization": f"Bearer {cfg['api_key']}", "Content-Type": "application/json", }, json={ "model": cfg["model_id"], "messages": [{"role": "user", "content": prompt}], "temperature": 0.3, }, timeout=cfg.get("timeout", 30), ) resp.raise_for_status() alert_text = resp.json()["choices"][0]["message"]["content"] for t in overdue_tasks: requests.post( t["webhook"], json={"msgtype": "text", "text": {"content": alert_text}}, timeout=10, )DAG 定义：
default_args = {"owner": "data-platform", "retries": 1, "retry_delay": timedelta(minutes=2)} with DAG( dag_id="timeliness_check", default_args=default_args, schedule_interval="*/10 * * * *", start_date=days_ago(1), catchup=False, tags=["monitor", "timeliness"], ) as dag: PythonOperator( task_id="check_timeliness", python_callable=check_timeliness, provide_context=True, )本地触发验证：把 DAG 文件放到$AIRFLOW_HOME/dags/，然后执行：
airflow dags list | grep timeliness_check airflow tasks test timeliness_check check_timeliness 2024-01-01airflow tasks test会直接运行函数并打印返回的 overdue_tasks 列表。如果配置表里有一条已经过期的记录，你会看到告警文本被打印出来，同时钉钉收到消息。成功结果类似：
[2024-01-01 02:00:00] {python.py:177} INFO - Done. Returned value was: [{'dag_id': 'daily_settlement', 'missing': ['settle_task'], 'deadline': '01:45', ...}]如果返回空列表，说明当前没有超时任务，链路是通的。这一步验证通过后，把 DAG 打开调度即可。
5. 常见报错排查：401、local proxy failed 与 reading choices
链路跑起来之前，大概率会遇到几个报错。这一节按真实报错对照排查，都是我在接入过程中实际碰到的。
第一个：401 Unauthorized。返回体通常是{"error": {"message": "Invalid API key"}}。原因有三种：Key 复制时带了空格、Key 已过期或被删除、请求头格式不对。检查Authorization头必须是Bearer sk-xxx，中间一个空格。如果你用 Airflow Connection，确认 Password 字段只填 Key，不要带Bearer前缀。另外确认 Base URL 是https://taotoken.net/api，不要多加/v1之外的路径。
第二个：local proxy failed或连接超时。这个报错通常出现在 requests 请求阶段，提示无法建立连接。排查顺序：先确认 Airflow worker 所在机器能访问https://taotoken.net/api，用curl -I https://taotoken.net/api测试；再检查环境变量里有没有残留的HTTP_PROXY、HTTPS_PROXY，如果有就 unset 掉；最后确认防火墙没有拦截出站 443。注意不要配置任何非官方的转发方式，直接用官方 API 地址即可。
第三个：reading 'choices'或KeyError: 'choices'。这个报错说明请求返回了 200，但响应体结构不是预期的 OpenAI 格式。常见原因是 Model ID 填错了，或者请求体里model字段为空。打印resp.json()看实际返回，如果是{"error": ...}说明模型不存在。去模型对话页面确认可用的 Model ID，填到配置文件的model_id字段。还有一种情况是响应被截断，检查timeout是否太短，调大到 30 秒以上。
第四个：OAuth相关报错，比如OAuth token expired。如果你用的是需要 OAuth 的接入方式，token 有有效期。但本篇用的是 API Key 方式，正常不会出现。如果出现，说明配置里混入了 OAuth 流程，检查是不是把别的工具的配置复制过来了。统一用 API Key 即可。
第五个：pymysql连接报__init__() takes 1 positional argument but 5 were given。这是位置参数写法导致的，改成关键字参数：
# 错误 db = pymysql.connect("localhost", "root", "xxx", "xxxx") # 正确 db = pymysql.connect(host="localhost", user="root", password="xxx", database="xxxx")第六个：DAG 导入报缩进错误IndentationError。Python 对缩进敏感，复制代码时容易混入 Tab 和空格。用python -m py_compile your_dag.py先本地编译检查，通过后再放到 dags 目录。
排查时建议打开 Airflow 的 task log，日志里会打印完整的请求和响应。如果告警没发出去但校验返回了 overdue_tasks，问题一定在 send_alert 函数，重点看 TaoToken 请求的返回码和 webhook 的返回体。
6. 把校验逻辑沉淀成可复用的监控资产
链路跑通之后，真正省事的地方在于扩展。新增一个监控任务，只需要往task_list插一行，指定 dag_id、task_id、最晚完成时间和通知组，不用改任何代码。换告警通道，只改notify_group里的 webhook。换模型或调整告警文案风格，只改taotoken.json里的 model_id 和 prompt。
几个实用技巧。第一，latest_complete_time建议比实际业务要求提前 5 到 10 分钟，给告警留出反应时间。第二，校验 DAG 的schedule_interval不要设太密，10 分钟一次足够，太密会给元数据库压力。第三，告警内容里带上 DAG 的 Airflow UI 链接，点一下就能跳过去看日志，这个字段可以从AIRFLOW__WEBSERVER__BASE_URL拼出来。第四，如果同一批任务频繁告警，考虑加一个静默期，比如同一 dag_id 两小时内只告警一次，避免刷屏。
如果你后续想把告警根因初判做得更细，比如让模型读 task log 给出可能原因，可以了解 Coding Plan 的额度方案，地址是 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 。接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ，里面有完整的接口说明和参数列表。API Keys 管理页面在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite ，需要轮换 Key 时从这里操作。
最后提醒一点：校验逻辑本身也要被监控。如果timeliness_check这个 DAG 自己挂了，所有告警都会静默。建议给它配一个 Airflow 自带的email_on_failure，或者用外部心跳检测。监控系统自己失联，是最难发现的问题。
