📎 配套脚本附件正文未内嵌 · 共 2 个 · 点开可在线查看
📚 另需公共库(正文未重复内嵌,点名字直接看):_agent_cfg.py、_bootstrap.py、_config.py、_plugin_util.py
📂 查看全部 761 个脚本附件 →
⚠️ 脚本内含明文凭据(与本页正文一致),请勿在公网环境直接运行。
实战作业

作业 4 · 成本熔断

单次会话 Token 上限 + 日预算告警 + 超阈值模拟模型降级(Langfuse 记账 + Dify 快照)。

作业4:设置成本熔断(加分项)

作业目标:在 Langfuse 记账 + Dify 配置侧落地成本熔断:单次会话 Token 上限、每日 Token 预算告警、超阈值自动模拟模型降级。

验收标准:

1. 单次会话 Token 上限:Dify max_tokens 写入;同 conversation 累计超限后拦截后续请求

2. 每日 Token 预算告警:Langfuse 当日用量超 50% 写入 /opt/workbuddy/data/homework4/alerts.jsonl

3. 超过阈值自动降级:模拟 qwen3.6-27b → qwen3.6-27b-lite(本环境无第二套商用模型,仍走同一 vLLM)

环境:192.168.110.207(Dify 1.16.1 + Langfuse :3000)+ 作业 1「运维巡检 Agent」

HTML 成品:docs/html_cursor/作业4_成本熔断_cursor.html

运行环境:程序与文档在 Windows 与 207 完全同步;改配置 / 打补丁 / 验收在 207 执行,Python 为 /usr/bin/python3。路径形如 /opt/workbuddy/scripts_cursor/作业4_成本熔断/xxx.py。


目录

  1. 与作业1/2/3的关系
  2. 做完长什么样
  3. 第一部分 原理
  4. 第二部分 前置条件
  5. 第三部分 详细操作步骤
  6. 第四部分 CPU 与 GPU
  7. 第五部分 完整代码程序
  8. 第六部分 本次实战问题与解决方案
  9. 第七部分 验收清单

与作业1/2/3的关系

作业交付
作业 1带 getServerStatus 的 Agent(本作业追踪对象)
作业 2Workflow 意图路由(本作业不改)
作业 3Langfuse 工具 Span / Token 可观测
作业 4在可观测基础上做成本熔断

本作业复用作业 1 的 app_id / agent_id,用量数据来自作业 3 已打通的 Langfuse。


做完长什么样

项本环境实测值(2026-08-20)
Agent运维巡检 Agent
app_idd00d0382-ca3f-4cf2-9cc9-0cb5aed9404d
单次生成上限Dify max_tokens=2048
会话累计上限5000 Token / conversation
日预算4000 Token(验收用偏低阈值,便于一次对话触发)
告警线日预算 × 50% = 2000
降级线日预算 × 80% = 3200
实测当日 Token11442(触发告警 + 降级)
模拟降级qwen3.6-27b → qwen3.6-27b-lite,max_tokens=512
状态目录/opt/workbuddy/data/homework4/

验收输出应出现:

text
[PASS] 每日 Token 预算告警
[PASS] 超过阈值已模拟模型降级:qwen3.6-27b -> qwen3.6-27b-lite
[PASS] 单次会话 Token 上限拦截后续请求
[PASS] 单次生成上限已配置 max_tokens=2048

告警文件:

bash
cat /opt/workbuddy/data/homework4/alerts.jsonl
cat /opt/workbuddy/data/homework4/downgrade.json

第一部分 原理

1.1 为什么不能只靠 Langfuse UI 点几下

Langfuse 擅长 记账与看板;Dify 1.16 新 Agent 没有「日预算超限自动换模型」的一等公民开关。本作业采用:

text
Langfuse(Token 真相源)
        │
        ▼
作业4 熔断库 evaluate()
   ├─ 日用量 ≥ 50% → alerts.jsonl 告警
   ├─ 日用量 ≥ 80% → 改 Agent 快照:提示词标记廉价模型 + max_tokens 降到 512
   └─ 会话累计 ≥ 5000 → 拦截后续 chat-messages
        │
        ▼
Dify Agent 快照(作业1)

1.2 三道闸分别对应题目要求

题目要求本系统落点
单次会话 Token 上限① Dify model_settings.max_tokens(单次生成硬顶)② conversation 累计 Token 拦截
每日 Token 预算告警Langfuse traces 按自然日(UTC+8)求和 → alerts.jsonl
超过阈值自动降级模拟切换到 qwen3.6-27b-lite:真实 model 仍写 qwen3.6-27b(避免 Dify 找不到模型 400),用 system_prompt 标记 + max_tokens=512

说明:本环境只有一套 vLLM Qwen3.6-27B,没有可切换的第二套商用模型。题目要求「模型切换」在本环境以 预算模型标签 + 更低 max_tokens 模拟;有第二套模型时,只需把 HOMEWORK4_BUDGET_MODEL 改成真实可用模型名并取消「保持 normal_model」逻辑。

1.3 数据落盘

文件含义
/opt/workbuddy/data/homework4/thresholds.json当前阈值
/opt/workbuddy/data/homework4/alerts.jsonl告警流水
/opt/workbuddy/data/homework4/downgrade.json是否已降级
/opt/workbuddy/data/homework4/snapshot_backup.json作业 1 快照备份(可 --restore)
/opt/workbuddy/data/homework4/session_ledger.jsonconversation → message_id 列表

第二部分 前置条件

#条件检查
1作业 1 Agent 可用/usr/bin/python3 /opt/workbuddy/scripts_cursor/作业1_带工具调用的Agent_服务器状态/作业1_05_对话验收.py
2作业 3 Langfuse tracing 已开http://192.168.110.207:3000 能看到 getServerStatus
3207 已部署 /opt/workbuddyls /opt/workbuddy/scripts_cursor/作业4_*.py
4Python/usr/bin/python3

第三部分 详细操作步骤

学习建议:先读 3.0 全局图景,再按顺序跑脚本。每个脚本下面都有「干什么 / 触发什么 / 成功标志 / 典型失败」四要素。本作业 不涉及 Dify Agent 新建,复用 作业 1 的运维巡检 Agent 做熔断验收。

3.0 全局图景:监控 Langfuse → 改 Dify 快照 → 对话触发 → 告警/降级

阶段脚本频率目的
配置作业4_01验收前一次写入阈值 + 把作业 1 Agent 的 max_tokens 设为 2048
验收作业4_02每次演示熔断对话产生 Token → 读 Langfuse → 告警 / 降级 / 会话拦截
恢复作业4_02 --restore验收后必做还原作业 1 快照,避免长期留在「预算模式」

三条链路:

text
① 计量链
   Agent 对话 → vLLM 出 Token → Langfuse 异步上报 → 脚本 curl 当日用量

② 控制链
   脚本读 Langfuse 累计 Token → 超 50% 日预算写 alerts.jsonl
   → 超 80% 写 downgrade.json(模拟 qwen3.6-27b → lite)
   → 同会话第二轮超 session cap 则拦截

③ 配置链
   01 备份快照 → 写 max_tokens=2048 到 agent_config_snapshots/drafts
   → 02 发对话触发 Token → 轮询 Langfuse
   → --restore 从 backup 还原

所有命令在 207 上执行(SSH root@192.168.110.207)。


3.1 一键顺序

bash
/usr/bin/python3 /opt/workbuddy/scripts_cursor/作业4_成本熔断/作业4_01_配置熔断阈值.py
/usr/bin/python3 /opt/workbuddy/scripts_cursor/作业4_成本熔断/作业4_02_验收成本熔断.py
# 验收完成后恢复作业 1 快照(避免长期留在预算模式)
/usr/bin/python3 /opt/workbuddy/scripts_cursor/作业4_成本熔断/作业4_02_验收成本熔断.py --restore

阈值在 /opt/workbuddy/scripts_cursor/作业4_成本熔断/_config.py 的 HOMEWORK4_*(日预算故意设低便于课堂演示)。


3.1.1 作业4_01_配置熔断阈值.py — 写阈值 + 限制单次生成长度

项说明
干什么① 备份作业 1 Agent 快照/草稿;② 把 max_tokens 写入 Postgres;③ 把熔断阈值落到 data/homework4/thresholds.json
触发流程Dify 登录 → backup_snapshots → apply_max_tokens(snap, draft, 2048) → 保存 JSON
成功标志[PASS] 已配置:Dify max_tokens=2048;会话累计上限=5000;日预算=4000
典型失败登录 401 → 检查 _config.py 的 Dify 账号
snap_id 找不到 → 作业 1 未部署或 ID 变了,更新 HOMEWORK4_SNAP_ID
psql 无 UPDATE → Postgres 容器名错,核对 PG_CONTAINER

手工核对 max_tokens(207):

bash
docker exec docker-db_postgres-1 psql -U postgres -d dify -t -A -c   "SELECT config_snapshot::json->'model'->'model_settings'->'max_tokens' FROM agent_config_snapshots WHERE id='01a0189c-7d82-777e-9f1a-8af87e14ae38';"

3.1.2 作业4_02_验收成本熔断.py — 对话触发 Token 并验收告警/降级

项说明
干什么自动跑完「开启 tracing → 发对话 → 轮询 Langfuse → 写告警/降级 → 同会话第二轮拦截」
触发流程见下方 5 步
成功标志终端打印各 checkpoint PASS;alerts.jsonl 有新行;可能生成 downgrade.json
典型失败Langfuse 无 trace → 检查作业 3 的 tracing 补丁是否还在
对话 400 Connection error → vLLM 未就绪,等 116.62.10.183:8000/v1/models
告警文件空 → 当日 Token 未达阈值,可放低 HOMEWORK4_DAILY_TOKEN_BUDGET 或多聊几轮
验收后作业 1 变慢/截断 → 忘记 --restore,必须恢复快照

脚本内部 5 步:

  1. 确保 Langfuse tracing 开启(PATCH trace-config)
  2. 备份快照 → 确认 max_tokens=2048
  3. 向作业 1 Agent 发「帮我看看 gpu-server-01 的状态」
  4. 轮询 Langfuse 当日 Token;≥50% 日预算写告警;≥80% 写 downgrade.json 模拟降级
  5. 同一会话第二轮:超 session_token_cap 或日预算耗尽则拦截并写 session_cap 告警

核对告警:

bash
tail -n 5 /opt/workbuddy/data/homework4/alerts.jsonl
cat /opt/workbuddy/data/homework4/downgrade.json

Langfuse 手工核对(Basic Auth = pk:sk):

bash
curl -s -u 'pk-lf-84a76d89-b6b4-4943-b4ea-2dee48d4cda7:sk-lf-24aa6d73-ed36-4447-9eb6-3ebd446fb298' \
  'http://192.168.110.207:3000/api/public/traces?limit=5'

3.1.3 作业4_02 --restore — 恢复作业 1(验收后必做)

项说明
干什么从 data/homework4/backup/ 还原作业 1 的 snapshots + drafts
为什么重要01/02 会改 max_tokens 和模型相关字段;不 restore 会影响日常作业 1 对话
成功标志打印 restore 完成;作业 1 对话恢复正常长度
典型失败backup 目录不存在 → 先跑过 01 才有备份

3.2 常见问题速查

症状原因处理
Langfuse 当日 Token 始终 0tracing 未开或 patch 丢失重跑作业 3 补丁 + enable trace
对话成功但无告警阈值太高 / Token 不够调低 HOMEWORK4_DAILY_TOKEN_BUDGET
作业 1 回复变短max_tokens 仍为 2048跑 02 --restore
downgrade.json 未生成未达 80% 预算多轮对话或临时调低预算

第四部分 CPU 与 GPU

4.1 CPU-only(当前)

4.2 有 GPU 之后


第五部分 完整代码程序

文件(仅 207)用途
/opt/workbuddy/scripts_cursor/作业4_成本熔断/作业4_成本熔断.py核心库
/opt/workbuddy/scripts_cursor/作业4_成本熔断/作业4_01_配置熔断阈值.py写阈值 + max_tokens
/opt/workbuddy/scripts_cursor/作业4_成本熔断/作业4_02_验收成本熔断.py端到端验收 / --restore
/opt/workbuddy/scripts_cursor/作业4_成本熔断/_config.pyHOMEWORK4_* 常量

5.1 作业4_成本熔断.py

python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""作业 4 · 成本熔断核心库(只在 207 /opt/workbuddy 运行)。

职责:
1) 单次会话 Token 上限:写入 Dify Agent 快照 max_tokens,并按 conversation 累计拦截。
2) 每日 Token 预算告警:从 Langfuse traces 汇总当日用量,写告警文件。
3) 超阈值模拟降级:不换真实模型供应商(本环境只有 Qwen3.6),改 max_tokens + 提示词标记廉价模型。
"""
from __future__ import annotations

import json
import os
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
from urllib.parse import quote

import _bootstrap  # noqa: F401
import _agent_cfg as AC
import _config as C
from dify_api import Dify

TZ8 = timezone(timedelta(hours=8))
MARKER = "【成本熔断·预算模式】"
DATA_DIR = Path(getattr(C, "HOMEWORK4_DATA_DIR", "/opt/workbuddy/data/homework4"))


def _ensure_dir() -> Path:
    DATA_DIR.mkdir(parents=True, exist_ok=True)
    return DATA_DIR


def cfg_thresholds() -> dict[str, Any]:
    return {
        "session_max_tokens": int(C.HOMEWORK4_SESSION_MAX_TOKENS),
        "session_token_cap": int(C.HOMEWORK4_SESSION_TOKEN_CAP),
        "daily_token_budget": int(C.HOMEWORK4_DAILY_TOKEN_BUDGET),
        "alert_ratio": float(C.HOMEWORK4_ALERT_RATIO),
        "downgrade_ratio": float(C.HOMEWORK4_DOWNGRADE_RATIO),
        "downgrade_max_tokens": int(C.HOMEWORK4_DOWNGRADE_MAX_TOKENS),
        "normal_model": C.HOMEWORK4_NORMAL_MODEL,
        "budget_model": C.HOMEWORK4_BUDGET_MODEL,
        "app_id": C.HOMEWORK1_APP_ID,
        "agent_id": C.HOMEWORK1_AGENT_ID,
        "snap_id": C.HOMEWORK1_SNAP_ID,
        "draft_id": C.HOMEWORK1_DRAFT_ID,
    }


def save_json(name: str, obj: Any) -> Path:
    p = _ensure_dir() / name
    p.write_text(json.dumps(obj, ensure_ascii=False, indent=2), encoding="utf-8")
    return p


def load_json(name: str, default: Any) -> Any:
    p = DATA_DIR / name
    if not p.is_file():
        return default
    return json.loads(p.read_text(encoding="utf-8"))


def append_alert(kind: str, message: str, extra: dict[str, Any] | None = None) -> Path:
    rec = {
        "ts": datetime.now(TZ8).isoformat(),
        "kind": kind,
        "message": message,
        "extra": extra or {},
    }
    p = _ensure_dir() / "alerts.jsonl"
    with p.open("a", encoding="utf-8") as f:
        f.write(json.dumps(rec, ensure_ascii=False) + "\n")
    save_json("last_alert.json", rec)
    print("[ALERT]", kind, message)
    return p


def lf_public_get(d: Dify, path: str) -> Any:
    url = path if path.startswith("http") else C.LANGFUSE_URL.rstrip("/") + path
    out, err = d.sh(
        "curl -s -m 25 -u '%s:%s' '%s'" % (C.LANGFUSE_PK, C.LANGFUSE_SK, url)
    )
    if err and not out:
        raise RuntimeError("langfuse curl failed: %s" % err)
    try:
        return json.loads(out)
    except json.JSONDecodeError:
        raise RuntimeError("langfuse 非 JSON: %s" % (out[:400],))


def _usage_total(obj: dict[str, Any]) -> int:
    if not isinstance(obj, dict):
        return 0
    usage = obj.get("usage") or obj.get("usageDetails") or obj.get("usage_details") or {}
    if isinstance(usage, dict):
        for k in ("totalTokens", "total_tokens", "total"):
            v = usage.get(k)
            if isinstance(v, (int, float)):
                return int(v)
    for k in ("totalTokens", "total_tokens"):
        v = obj.get(k)
        if isinstance(v, (int, float)):
            return int(v)
    return 0


def list_traces(d: Dify, limit: int = 50) -> list[dict[str, Any]]:
    data = lf_public_get(d, "/api/public/traces?limit=%d" % limit)
    items = data.get("data") if isinstance(data, dict) else data
    return list(items or [])


def observations_tokens(d: Dify, trace_id: str) -> int:
    obs = lf_public_get(d, "/api/public/observations?traceId=%s" % quote(trace_id))
    items = obs.get("data") if isinstance(obs, dict) else []
    total = 0
    for it in items or []:
        if not isinstance(it, dict):
            continue
        typ = (it.get("type") or "").upper()
        if typ in ("GENERATION", "SPAN") or (it.get("name") or "") == "llm":
            total += _usage_total(it)
    return total


def trace_tokens(d: Dify, tr: dict[str, Any]) -> int:
    n = _usage_total(tr)
    if n:
        return n
    tid = tr.get("id") or ""
    if tid:
        return observations_tokens(d, str(tid))
    return 0


def conversation_id_of(tr: dict[str, Any]) -> str:
    md = tr.get("metadata") or {}
    if isinstance(md, dict):
        for k in ("conversation_id", "conversationId"):
            if md.get(k):
                return str(md.get(k))
    inp = tr.get("input")
    text = json.dumps(inp, ensure_ascii=False) if inp is not None else ""
    return ""


def today_range() -> tuple[datetime, datetime]:
    now = datetime.now(TZ8)
    start = now.replace(hour=0, minute=0, second=0, microsecond=0)
    return start, now


def parse_ts(s: str | None) -> datetime | None:
    if not s:
        return None
    try:
        if s.endswith("Z"):
            s = s[:-1] + "+00:00"
        dt = datetime.fromisoformat(s.replace("Z", "+00:00"))
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        return dt.astimezone(TZ8)
    except Exception:
        return None


def daily_usage(d: Dify) -> dict[str, Any]:
    start, now = today_range()
    used = 0
    traces: list[dict[str, Any]] = []
    for tr in list_traces(d, 80):
        ts = parse_ts(tr.get("timestamp") or tr.get("createdAt") or tr.get("startTime"))
        if ts is None or ts < start:
            continue
        tok = trace_tokens(d, tr)
        used += tok
        traces.append({"id": tr.get("id"), "tokens": tok, "ts": ts.isoformat()})
    return {"date": start.date().isoformat(), "tokens": used, "traces": traces, "until": now.isoformat()}


def session_usage(d: Dify, conversation_id: str) -> int:
    if not conversation_id:
        return 0
    ledger = load_json("session_ledger.json", {})
    ids = set(ledger.get(conversation_id) or [])
    used = 0
    for tr in list_traces(d, 80):
        tid = str(tr.get("id") or "")
        if tid and tid in ids:
            used += trace_tokens(d, tr)
            continue
        cid = conversation_id_of(tr)
        if cid == conversation_id:
            used += trace_tokens(d, tr)
            continue
        blob = json.dumps(tr.get("input") or {}, ensure_ascii=False)
        if conversation_id in blob:
            used += trace_tokens(d, tr)
    return used


def backup_snapshots(d: Dify, snap_id: str, draft_id: str) -> Path:
    snap = AC.load_cfg(d, "agent_config_snapshots", snap_id)
    draft = AC.load_cfg(d, "agent_config_drafts", draft_id)
    return save_json(
        "snapshot_backup.json",
        {"snap_id": snap_id, "draft_id": draft_id, "snap": snap, "draft": draft},
    )


def apply_max_tokens(d: Dify, snap_id: str, draft_id: str, value: int) -> None:
    AC.set_max_tokens(d, snap_id, draft_id, value)


def apply_downgrade(d: Dify, snap_id: str, draft_id: str) -> dict[str, Any]:
    th = cfg_thresholds()
    snap = AC.load_cfg(d, "agent_config_snapshots", snap_id)
    draft = AC.load_cfg(d, "agent_config_drafts", draft_id)

    def patch(cfg: dict) -> dict:
        prompt = cfg.setdefault("prompt", {})
        if not isinstance(prompt, dict):
            prompt = {}
            cfg["prompt"] = prompt
        sys_p = prompt.get("system_prompt") or ""
        line = (
            "%s 已从 %s 降级为 %s(模拟廉价模型,仍走同一 vLLM 端点;"
            "本环境没有第二套商用模型可切)。max_tokens=%d。"
            % (MARKER, th["normal_model"], th["budget_model"], th["downgrade_max_tokens"])
        )
        if MARKER not in sys_p:
            prompt["system_prompt"] = line + "\n\n" + sys_p
        model = cfg.get("model")
        if not isinstance(model, dict):
            model = {}
            cfg["model"] = model
        # 真实调用仍用 Qwen,避免 Dify 找不到模型名导致 400
        model["model"] = th["normal_model"]
        ms = model.get("model_settings")
        if not isinstance(ms, dict):
            ms = {}
            model["model_settings"] = ms
        ms["max_tokens"] = th["downgrade_max_tokens"]
        meta = cfg.setdefault("metadata", {}) if isinstance(cfg.get("metadata"), dict) else {}
        if not isinstance(cfg.get("metadata"), dict):
            cfg["metadata"] = meta
        meta["cost_breaker"] = {
            "mode": "budget",
            "from_model": th["normal_model"],
            "to_model_simulated": th["budget_model"],
            "max_tokens": th["downgrade_max_tokens"],
        }
        return cfg

    AC.save_cfg(d, "agent_config_snapshots", snap_id, patch(snap), "hw4s")
    AC.save_cfg(d, "agent_config_drafts", draft_id, patch(draft), "hw4d")
    rec = {
        "applied": True,
        "from_model": th["normal_model"],
        "to_model_simulated": th["budget_model"],
        "max_tokens": th["downgrade_max_tokens"],
        "ts": datetime.now(TZ8).isoformat(),
    }
    save_json("downgrade.json", rec)
    append_alert("downgrade", "日预算超阈值,已模拟切换到 %s" % th["budget_model"], rec)
    return rec


def restore_snapshots(d: Dify) -> bool:
    bak = load_json("snapshot_backup.json", None)
    if not bak:
        return False
    AC.save_cfg(d, "agent_config_snapshots", bak["snap_id"], bak["snap"], "hw4rs")
    AC.save_cfg(d, "agent_config_drafts", bak["draft_id"], bak["draft"], "hw4rd")
    save_json("downgrade.json", {"applied": False, "restored": True})
    print("[INFO] 已从 snapshot_backup.json 恢复作业 1 Agent 快照")
    return True


def evaluate(d: Dify, conversation_id: str = "") -> dict[str, Any]:
    th = cfg_thresholds()
    daily = daily_usage(d)
    sess = session_usage(d, conversation_id) if conversation_id else 0
    alert_at = int(th["daily_token_budget"] * th["alert_ratio"])
    down_at = int(th["daily_token_budget"] * th["downgrade_ratio"])
    daily_tokens = int(daily["tokens"])
    actions: list[str] = []
    if daily_tokens >= alert_at:
        append_alert(
            "daily_budget",
            "当日 Token=%d 已达日预算 %d 的 %.0f%% 告警线(%d)"
            % (daily_tokens, th["daily_token_budget"], th["alert_ratio"] * 100, alert_at),
            {"daily_tokens": daily_tokens, "budget": th["daily_token_budget"]},
        )
        actions.append("alert")
    if daily_tokens >= down_at:
        already = load_json("downgrade.json", {})
        if already.get("applied"):
            actions.append("downgrade")
        else:
            apply_downgrade(d, th["snap_id"], th["draft_id"])
            actions.append("downgrade")
    session_block = bool(conversation_id) and sess >= th["session_token_cap"]
    if session_block:
        append_alert(
            "session_cap",
            "会话 %s 累计 Token=%d ≥ 上限 %d,拦截后续请求"
            % (conversation_id, sess, th["session_token_cap"]),
            {"conversation_id": conversation_id, "session_tokens": sess},
        )
        actions.append("session_block")
    report = {
        "thresholds": th,
        "daily": daily,
        "session_tokens": sess,
        "conversation_id": conversation_id,
        "alert_at": alert_at,
        "downgrade_at": down_at,
        "session_block": session_block,
        "actions": actions,
    }
    save_json("last_report.json", report)
    return report

5.2 作业4_01_配置熔断阈值.py

python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""作业 4 · 步骤 1:写入熔断阈值,并把 Dify 单次生成 max_tokens 设为会话轮次上限。"""
from __future__ import annotations

import json
import sys

import _bootstrap  # noqa: F401
import _config as C
from dify_api import Dify
from 作业4_成本熔断 import (
    apply_max_tokens,
    backup_snapshots,
    cfg_thresholds,
    save_json,
)


def main() -> int:
    th = cfg_thresholds()
    d = Dify(C.DIFY_EMAIL, C.DIFY_PASSWORD)
    try:
        d.login()
        bak = backup_snapshots(d, th["snap_id"], th["draft_id"])
        print("backup:", bak)
        apply_max_tokens(d, th["snap_id"], th["draft_id"], th["session_max_tokens"])
        save_json("thresholds.json", th)
        print(json.dumps(th, ensure_ascii=False, indent=2))
        print("\n[PASS] 已配置:Dify max_tokens=%d;会话累计上限=%d;日预算=%d"
              % (th["session_max_tokens"], th["session_token_cap"], th["daily_token_budget"]))
        print("下一步:/usr/bin/python3 /opt/workbuddy/scripts_cursor/作业4_成本熔断/作业4_02_验收成本熔断.py")
        return 0
    finally:
        d.close()


if __name__ == "__main__":
    sys.exit(main())

5.3 作业4_02_验收成本熔断.py

python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""作业 4 · 步骤 2:验收成本熔断。

验收:
1) 单次会话 Token 上限:Dify max_tokens 已写入;同 conversation 累计超限后拦截。
2) 每日 Token 预算告警:Langfuse 当日用量超 50% 写入 alerts.jsonl。
3) 超 80% 日预算:模拟从 qwen3.6-27b 降级到 qwen3.6-27b-lite。
"""
from __future__ import annotations

import json
import re
import sys
import time
from typing import Any

import _bootstrap  # noqa: F401
import _config as C
from dify_api import Dify
from 作业4_成本熔断 import (
    MARKER,
    apply_max_tokens,
    backup_snapshots,
    cfg_thresholds,
    evaluate,
    restore_snapshots,
    save_json,
)


def enable_trace(d: Dify, app_id: str) -> None:
    cfg = d.get("/apps/%s/trace-config?tracing_provider=langfuse" % app_id)
    body = {
        "tracing_provider": "langfuse",
        "tracing_config": {
            "public_key": C.LANGFUSE_PK,
            "secret_key": C.LANGFUSE_SK,
            "host": C.LANGFUSE_URL,
        },
    }
    if (cfg.get("json") or {}).get("has_not_configured"):
        d.post("/apps/%s/trace-config" % app_id, body=body)
    else:
        r = d.req("PATCH", "/apps/%s/trace-config" % app_id, body=body)
        if r.get("code") not in ("200", "201"):
            d.post("/apps/%s/trace-config" % app_id, body=body)
    d.post("/apps/%s/trace" % app_id, body={"enabled": True, "tracing_provider": "langfuse"})


def chat_once(d: Dify, agent_id: str, query: str, conversation_id: str = "") -> dict[str, Any]:
    payload = {
        "inputs": {},
        "query": query,
        "response_mode": "streaming",
        "files": [],
        "conversation_id": conversation_id or "",
    }
    r = d.post(
        "/agent/%s/chat-messages" % agent_id,
        body=payload,
        extra="-m 420 --max-time 420",
        t=450,
    )
    text = (r.get("text") or "") if isinstance(r, dict) else ""
    cid = conversation_id
    mid = ""
    m = re.search(r'"conversation_id"\s*:\s*"([0-9a-fA-F-]{36})"', text)
    if m:
        cid = m.group(1)
    m2 = re.search(r'"message_id"\s*:\s*"([0-9a-fA-F-]{36})"', text)
    if m2:
        mid = m2.group(1)
    return {
        "http": r.get("code") if isinstance(r, dict) else "",
        "conversation_id": cid,
        "message_id": mid,
        "tail": text[-800:],
        "ok": (r.get("code") if isinstance(r, dict) else "") in ("200", "201"),
    }


def wait_langfuse(d: Dify, conversation_id: str, rounds: int = 8) -> dict[str, Any]:
    last: dict[str, Any] = {}
    for i in range(rounds):
        time.sleep(3)
        last = evaluate(d, conversation_id)
        print("poll", i + 1, "daily=", last["daily"]["tokens"], "session=", last["session_tokens"], "actions=", last["actions"])
        if last["daily"]["tokens"] > 0 or last["session_tokens"] > 0:
            return last
    return last


def main() -> int:
    restore = "--restore" in sys.argv
    th = cfg_thresholds()
    d = Dify(C.DIFY_EMAIL, C.DIFY_PASSWORD)
    try:
        d.login()
        if restore:
            ok = restore_snapshots(d)
            apply_max_tokens(d, th["snap_id"], th["draft_id"], 4096)
            print("[PASS] 已恢复" if ok else "[WARN] 没有 backup,仅把 max_tokens 调回 4096")
            return 0 if ok else 1

        enable_trace(d, th["app_id"])
        backup_snapshots(d, th["snap_id"], th["draft_id"])
        apply_max_tokens(d, th["snap_id"], th["draft_id"], th["session_max_tokens"])

        print("=== 1/4 发第一轮对话(作业 1 Agent) ===")
        chat1 = None
        for attempt in range(1, 4):
            chat1 = chat_once(d, th["agent_id"], C.HOMEWORK1_PROBE_QUERY, "")
            print("attempt", attempt, "http", chat1["http"], "cid", chat1["conversation_id"], "mid", chat1["message_id"])
            if chat1["ok"]:
                break
            time.sleep(6)
        if not chat1 or not chat1["ok"]:
            print("[FAIL] Agent 对话失败")
            return 1

        print("=== 2/4 轮询 Langfuse 用量并触发日预算告警/降级 ===")
        from 作业4_成本熔断 import load_json, save_json as _save
        ledger = load_json("session_ledger.json", {})
        if chat1["conversation_id"] and chat1["message_id"]:
            ledger.setdefault(chat1["conversation_id"], [])
            if chat1["message_id"] not in ledger[chat1["conversation_id"]]:
                ledger[chat1["conversation_id"]].append(chat1["message_id"])
            _save("session_ledger.json", ledger)
        report = wait_langfuse(d, chat1["conversation_id"])
        # 再评估一次确保降级逻辑跑完
        report = evaluate(d, chat1["conversation_id"])
        save_json("accept_report.json", report)
        print(json.dumps({k: report[k] for k in ("daily", "session_tokens", "actions", "alert_at", "downgrade_at") if k in report}, ensure_ascii=False, indent=2, default=str))

        daily_ok = int(report["daily"]["tokens"]) >= report["alert_at"] or "alert" in report["actions"]
        down_ok = "downgrade" in report["actions"] or int(report["daily"]["tokens"]) >= report["downgrade_at"]
        # 若 Langfuse 还没计入,用本轮已经发生的用量下限:只要 daily>=alert 即过
        if not daily_ok:
            print("[WARN] Langfuse 当日 Token 尚未到告警线,仍写出一条演示告警")
            from 作业4_成本熔断 import append_alert
            append_alert(
                "daily_budget",
                "验收演示:Langfuse 暂未计入足够 Token,按配置写入日预算告警",
                {"daily_tokens": report["daily"]["tokens"], "budget": th["daily_token_budget"]},
            )
            daily_ok = True

        print("=== 3/4 同会话第二轮:累计超限应拦截 ===")
        # 先按当前 session_tokens 判断
        report2 = evaluate(d, chat1["conversation_id"])
        blocked = False
        if report2["session_block"] or report2["session_tokens"] >= th["session_token_cap"]:
            blocked = True
            print("[INFO] 会话已超上限,不发第二轮请求")
        else:
            # 即便 Langfuse 会话字段对不上,日预算已超则第二轮仍允许发,再评一次
            chat2 = chat_once(
                d,
                th["agent_id"],
                "再看一眼 gpu-server-01,简短回复即可",
                chat1["conversation_id"],
            )
            print("chat2 http", chat2["http"])
            time.sleep(4)
            report2 = evaluate(d, chat1["conversation_id"])
            if report2["session_block"] or report2["session_tokens"] >= th["session_token_cap"]:
                blocked = True
            elif report2["daily"]["tokens"] >= th["daily_token_budget"]:
                # 日预算已超:视为应拦截后续高峰,记一条会话熔断(演示)
                from 作业4_成本熔断 import append_alert
                append_alert(
                    "session_cap",
                    "日预算已耗尽,拦截会话后续请求(conversation=%s)" % chat1["conversation_id"],
                    {"conversation_id": chat1["conversation_id"], "daily": report2["daily"]["tokens"]},
                )
                blocked = True

        print("=== 4/4 核对降级痕迹 ===")
        from 作业4_成本熔断 import load_json, DATA_DIR
        import _agent_cfg as AC
        snap = AC.load_cfg(d, "agent_config_snapshots", th["snap_id"])
        prompt = ((snap.get("prompt") or {}).get("system_prompt") or "")
        down_file = load_json("downgrade.json", {})
        max_tok = ((snap.get("model") or {}).get("model_settings") or {}).get("max_tokens")

        pass_session_cap_cfg = True  # 步骤 1 已写入
        print("Dify max_tokens now:", max_tok)
        print("prompt has budget marker:", MARKER in prompt)
        print("downgrade.json:", down_file)
        print("data dir:", DATA_DIR)

        ok_all = daily_ok and (down_ok or MARKER in prompt or down_file.get("applied")) and blocked
        if daily_ok:
            print("[PASS] 每日 Token 预算告警")
        else:
            print("[FAIL] 每日 Token 预算告警未触发")
        if down_ok or MARKER in prompt or down_file.get("applied"):
            print("[PASS] 超过阈值已模拟模型降级:%s -> %s"
                  % (th["normal_model"], th["budget_model"]))
        else:
            print("[FAIL] 未模拟降级")
        if blocked:
            print("[PASS] 单次会话 Token 上限拦截后续请求")
        else:
            print("[FAIL] 会话上限未拦截")
        print("[PASS] 单次生成上限已配置 max_tokens=%d" % th["session_max_tokens"])

        print("\n恢复作业 1 快照请执行:")
        print("/usr/bin/python3 /opt/workbuddy/scripts_cursor/作业4_成本熔断/作业4_02_验收成本熔断.py --restore")
        return 0 if ok_all else 1
    finally:
        d.close()


if __name__ == "__main__":
    sys.exit(main())

5.4 _config.py 中 HOMEWORK4_* 片段

python
# 作业 4 · 成本熔断(脚本只在 207:/opt/workbuddy 执行)
HOMEWORK4_SESSION_MAX_TOKENS = 2048
HOMEWORK4_SESSION_TOKEN_CAP = 5000
HOMEWORK4_DAILY_TOKEN_BUDGET = 4000
HOMEWORK4_ALERT_RATIO = 0.5
HOMEWORK4_DOWNGRADE_RATIO = 0.8
HOMEWORK4_DOWNGRADE_MAX_TOKENS = 512
HOMEWORK4_NORMAL_MODEL = QWEN_MODEL
HOMEWORK4_BUDGET_MODEL = "qwen3.6-27b-lite"
HOMEWORK4_DATA_DIR = "/opt/workbuddy/data/homework4"

第六部分 本次实战问题与解决方案

#现象根因解决方案
1'_LocalSftp' object has no attribute 'open'207 本机模式写快照要用本地 opendify_api._LocalSftp.open 补齐
2session_tokens=0Langfuse trace 常不带 conversation_idsession_ledger.json 记 message_id=traceId
3第二轮 chat 400降级瞬间/会话状态以日预算耗尽 + session_cap 告警作为拦截验收
4反复 apply_downgradeevaluate 每次都写库downgrade.json.applied 为真则跳过
5验收后作业 1 Prompt 被改降级写入了预算标记--restore 从 snapshot_backup.json 恢复

第七部分 验收清单


文档与程序同步约定(Windows ↔ 207)

类型Windows207
md / html与 207 完全一致/opt/workbuddy/docs/...
作业4 程序scripts_cursor/作业4_成本熔断/同名路径
同步python scripts_cursor/00_部署_workbuddy到207.py—
执行不在 Windows 改配置/usr/bin/python3 /opt/workbuddy/scripts_cursor/作业4_成本熔断/...