JellyToken 在自动化工作流中的调用与异常处理

要把 JellyToken 接进 CI/CD 或定时任务，核心不是“调通一次”，而是把鉴权、参数校验、错误分支和重跑入口都放进同一个封装里，让任务在异常后能留下日志并保持可介入。下面这套处理方式不依赖特定平台，适用场景是：你有一个需要周期运行的 Python 脚本，JellyToken 负责生成文本结果，脚本把结果写入文件或发送到内部系统。操作动作是：先定义输入输出，再封装调用函数，然后分类捕获异常，接上日志和 webhook 告警，最后给关键任务留出手动重跑或降级路径。验证方式：用一条故意写错密钥的请求和一条断网请求，分别确认日志记录和错误分类正确；风险边界是：JellyToken 服务本身的超时策略和限流字段可能随版本变化，需要在接入前用实际账号做一次连通性测试。

如果目标是让自动化任务在JellyToken不稳定时仍可控，建议把“调用”与“业务逻辑”彻底分开：一个模块只负责发请求、解析响应、抛分类异常；另一个模块负责记录日志、发通知、决定是否重跑。这样可以先做到异常可见，再说要不要自动重试。若任务不是强实时，优先用手动重跑或降级预案，避免在无充分依据时盲目调大重试次数。每项异常处理都要能在日志里看到最终结果与处置动作，边界在于：JellyToken的接口字段、错误码定义和限流策略需结合真实环境确认，本文不针对具体平台缺陷做推断。

定义自动化任务需求

先别急着写调用代码。定一个具体任务示例：每天凌晨 2 点，脚本读取昨天生成的测试报告 JSON，把“失败用例数、关键错误摘要、涉及模块”发给 JellyToken，要求生成 300 字以内的中文摘要，再把摘要写入指定 Markdown 文件。这个任务的输入是报告 JSON，输出是文本摘要，失败时的处置是“保留原始报告并发送告警”。

输入输出定清楚后，再决定调用逻辑：一次任务只处理一个报告文件；JellyToken 请求超时设为 30 秒；如果报告中没有失败用例，脚本直接跳过调用并记录“无需生成”。这样做的好处是，自动化脚本可测试、可重跑，且不会因为空数据浪费一次 API 调用。

封装统一的调用函数

把 JellyToken 的 HTTP 请求包成一个独立函数，参数包括提示词、模型名、超时时间和温度等常见参数，返回解析后的文本。不要在每个业务脚本里重复写 requests 逻辑，否则一旦需要调整鉴权方式或超时策略，就要改多处。以下是一个可替换配置的接入骨架：

import requests
import json
from typing import Optional

def call_jellytoken(
    prompt: str,
    model: str = "default",
    temperature: float = 0.2,
    max_tokens: int = 500,
    timeout: int = 30,
    api_key: Optional[str] = None,
) -> str:
    """调用 JellyToken，返回解析后的文本。

    实际 URL、请求体字段和鉴权头需要按自己账号的接口文档替换。
    """
    if not api_key:
        raise ValueError("缺少 api_key")

    url = "https://your-endpoint.example.com/v1/chat/completions"  # 替换为实际地址

    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }

    payload = {
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "temperature": temperature,
        "max_tokens": max_tokens,
    }

    resp = requests.post(url, headers=headers, json=payload, timeout=timeout)

    # 如果 HTTP 状态码不是 2xx，由调用方负责分类和处理
    resp.raise_for_status()

    data = resp.json()

    # 假设响应中内容位于 choices[0].message.content，需按实际结构替换
    try:
        return data["choices"][0]["message"]["content"]
    except (KeyError, IndexError, TypeError) as exc:
        raise ValueError(f"无法从响应中提取文本: {data}") from exc

函数里已经做了响应结构校验，解析失败时会抛 ValueError。这样调用方只需关心 call_jellytoken(...) 的返回文本或异常，不需要关心请求细节。建议把 api_key 放在环境变量或密钥管理服务里，不要写死在脚本中。

捕获并分类异常

自动化任务会遇到的异常大致分三类：网络连接类、认证类、API 逻辑类。分别为它们定义异常子类，再写一个统一的捕获入口。下面给出异常类定义与捕获逻辑：

class JellyTokenError(Exception):
    """JellyToken 相关错误基类"""

class JellyTokenNetworkError(JellyTokenError):
    """网络连接、超时、DNS 解析问题"""

class JellyTokenAuthError(JellyTokenError):
    """API key 无效、权限不足"""

class JellyTokenAPIError(JellyTokenError):
    """请求被 API 拒绝，例如参数错误、资源限制、模型不存在"""


def classify_error(exc: Exception, resp: Optional[requests.Response] = None) -> JellyTokenError:
    """把 requests 异常和响应状态码转换为 JellyToken 异常"""
    if isinstance(exc, requests.exceptions.Timeout):
        return JellyTokenNetworkError("请求超时")
    if isinstance(exc, requests.exceptions.ConnectionError):
        return JellyTokenNetworkError("网络连接失败")

    if resp is not None:
        status = resp.status_code
        if status == 401 or status == 403:
            return JellyTokenAuthError(f"认证失败，状态码: {status}")
        if status >= 400:
            return JellyTokenAPIError(f"API 错误，状态码: {status}，响应: {resp.text[:200]}")

    return JellyTokenError(f"未分类异常: {exc}")

在业务代码里，捕获时先判断异常类型，再决定要继续、重试还是终止：

try:
    result = call_jellytoken(prompt, api_key=api_key)
except JellyTokenAuthError:
    # 认证失败不会自己好，需要人工更换密钥，直接发告警并终止任务
    raise
except JellyTokenNetworkError:
    # 网络问题可以稍后重试，但先记录日志
    print("网络异常，稍后重试")
    # 这里可以执行主动重试，也可以交给外部调度器在下一个周期再跑
except JellyTokenAPIError as exc:
    # 如果错误是“内容合规”或“模型不存在”，重试也没有意义
    print(exc)
except Exception as exc:
    # 未知异常，保留堆栈
    print(f"未预期错误: {exc}")

注意：不要对所有异常统一执行重试。401 认证错误重试多少次都会失败，反而可能触发账号风控；API 层错误里若是参数格式问题，重试也没有意义。只有网络超时和 5xx 类服务端错误才适合主动重试，且重试次数建议不超过两次，每次间隔至少 30 秒。

集成日志与告警

日志的目的是让维护者知道“什么时候、哪个环节失败、最终怎么处置”。建议使用 Python 标准库 logging，将异常写入文件，同时通过 webhook 发送通知。以下是一个可直接替换 API 地址和密钥的示例：

import logging
import requests

logging.basicConfig(
    filename="jellytoken_task.log",
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)

def send_webhook(message: str, webhook_url: str) -> None:
    """通过 webhook 发送文本，适用于钉钉/企业微信/自建接口。"""
    try:
        requests.post(webhook_url, json={"text": message}, timeout=10)
    except requests.exceptions.RequestException as exc:
        logging.error("webhook 发送失败: %s", exc)

def run_task(prompt: str):
    webhook_url = "https://your-webhook.example.com/notify"  # 替换
    try:
        result = call_jellytoken(prompt, api_key="your-env-key")
    except JellyTokenAuthError as exc:
        logging.critical("认证失败，终止任务: %s", exc)
        send_webhook(f"JellyToken 任务认证失败: {exc}", webhook_url)
        raise
    except JellyTokenNetworkError as exc:
        logging.warning("网络异常: %s，将保留任务现场", exc)
        send_webhook(f"JellyToken 网络异常: {exc}", webhook_url)
    except JellyTokenAPIError as exc:
        logging.error("API 错误: %s", exc)
    else:
        logging.info("任务成功，结果长度: %d", len(result))
        return result

日志里至少要记录：任务名称、触发时间、异常类别、异常详情、重试次数（如果有）。不要只记录一行“调用失败”，否则无法判断是密钥问题还是网络问题。webhook 通知应只发给真正需要处理的异常，例如认证失败和连续重试失败；单次网络闪断可以只记录日志，不用打扰值班人。

设置失败补偿机制

补偿机制不是指无限自动重试，而是让关键任务在失败后仍有机会完成。对于定时生成测试报告摘要这类任务，推荐两种方式：手动重跑脚本和降级预案。

手动重跑的前提是脚本幂等，即每次运行时输入输出都不受前一次运行状态影响。撰写任务脚本时，让程序接受一个日期参数：

python generate_summary.py `--date` 2026-02-14

这样某一天失败后，维护者只需重新执行同一条命令，脚本会重新读取那个日期的测试报告并生成摘要，不会因为“上次已经写了部分结果”而冲突。若 JellyToken 连续异常且不能再等，可以降级为“直接摘取测试报告中的失败用例列表”，生成一个无修饰的纯文本摘要写入目标文件，并在第一行标注“降级摘要”。这样下游消费者依然能获得数据，只是没有 AI 润色。

可以把补偿逻辑定义为：网络错误重试 2 次，每次间隔 60 秒；若仍失败，写入标记文件 jellytoken_retry_next_run.flag，由外部的 cron 或 CI 调度器在下一个周期看到标记后自动重跑。但需要结合你的实际调度器能力，确认标记文件机制是否可靠。如果调度器只认任务退出码，建议让脚本在最终失败时返回非 0 退出码，再由上层的 CI 平台触发重试。不要把所有异常都吞掉后返回 0，否则任务失败了系统还以为一切正常。

任何补偿措施都必须在日志里留下可追踪的记录，包括发生了什么异常、采取了哪种补偿路径、最终是否成功。这样才能在后续优化重试策略时找到依据。