HiveGPT AI 学习
登录
H · AI 应用开发进阶第 8 课⏱ 20 分钟

生产化：可观测、缓存、模型路由和成本

学完你能
  • 为每个请求记录 trace 和 span，看懂错误率、p95 延迟、费用和工具失败率
  • 用精确缓存、提示词缓存和模型路由降低成本，出错时能回退
  • 给提示词和模型标上版本，跑过评估再灰度上线，随时能回滚
Yui和Kai协作管理可观测的AI应用生产流水线
Yui和Kai协作管理可观测的AI应用生产流水线AI 生成配图

A8 讲了上线的基本功：估算成本、错误重试、保护 Key。应用真有人用以后，问题会变成：用户说「刚才那次回答又慢又错」，你能找到那次请求吗？这周费用涨了一倍，是哪个功能、哪个模型花的？换了新提示词，到底变好还是变坏了？ 这一课把这些补上。

链路追踪：一次请求，一个 trace ​

Yui和Kai沿请求轨迹检查模型与工具调用
Yui和Kai沿请求轨迹检查模型与工具调用AI 生成配图

一个用户请求里可能有好几次模型调用和工具调用（H5 的工作流、H7 的 Agent）。给每个请求生成一个 trace id，请求里的每一次模型调用或工具调用记一条 span：

字段内容
trace_id同一请求的所有 span 共用
span这一步叫什么：route、retrieve、answer、tool:search_docs
model、版本gpt-5.5、提示词 support-v3
token输入、输出、缓存命中的输入
耗时、费用毫秒；按单价算出的金额
错误异常类型和信息，没出错为空
python
import json, os, re, time, uuid
from openai import OpenAI

client = OpenAI(base_url="https://hivegpt.cn/v1", api_key=os.environ["HIVEGPT_API_KEY"], timeout=60)

# 每百万 token 的输入 / 输出单价（元）。示意数字，请按 HiveGPT 模型列表里的实际单价填写
PRICE = {"gpt-5.5": (10.0, 40.0), "gpt-5.4-mini": (1.0, 4.0)}

def redact(text):
    text = re.sub(r"\d{17}[\dXx]", "[证件号]", text)
    text = re.sub(r"1[3-9]\d{9}", "[手机号]", text)
    return re.sub(r"[\w.+-]+@[\w-]+\.[\w.-]+", "[邮箱]", text)

def write_span(span):
    with open("traces.jsonl", "a", encoding="utf-8") as f:   # 正式环境写进日志系统或数据库
        f.write(json.dumps(span, ensure_ascii=False) + "\n")

def traced_chat(trace_id, span_name, **kwargs):
    span = {"trace_id": trace_id, "span": span_name, "model": kwargs["model"], "error": None,
            "input": redact(str(kwargs["messages"][-1]["content"]))[:300]}   # 只存脱敏、截断后的片段
    t0 = time.time()
    try:
        resp = client.chat.completions.create(**kwargs)
        u = resp.usage
        details = getattr(u, "prompt_tokens_details", None)
        p_in, p_out = PRICE.get(kwargs["model"], (0, 0))
        span.update(input_tokens=u.prompt_tokens, output_tokens=u.completion_tokens,
                    cached_tokens=(details.cached_tokens or 0) if details else 0,
                    cost=round((u.prompt_tokens * p_in + u.completion_tokens * p_out) / 1e6, 6))
        return resp
    except Exception as e:
        span["error"] = f"{type(e).__name__}: {e}"
        raise
    finally:
        span["ms"] = int((time.time() - t0) * 1000)
        write_span(span)

trace_id = uuid.uuid4().hex     # 每个用户请求生成一个，和回答一起返回给前端

工具调用也用 write_span 记一条（span 写 tool:工具名，记耗时和是否失败）。量大了以后可以换成 OpenTelemetry 这类标准方案，字段思路一样。

存提示词要小心：

  • 写入前脱敏手机号、邮箱、证件号；不要记录 API Key 和请求头。
  • 默认只存截断后的片段；确实需要全文排查时，放在权限更严、保存期更短的地方，到期删除。
  • 在隐私说明里告诉用户你会记录什么。

看板和告警 ​

把 span 按时间汇总，至少看这几项：

指标怎么算告警示例
错误率出错的请求 ÷ 总请求，按 5 分钟统计超过 2% 并持续 10 分钟
p95 延迟95% 的请求在这个时间内完成比上周同一时段高 50%
每日费用所有 span 的 cost 之和超过日预算的 80%
工具失败率失败的工具调用 ÷ 工具调用总数某个工具超过 10%
  • 看 p95，不只看平均：平均 2 秒，可能掩盖了一成请求要等 15 秒。
  • 按模型、功能、提示词版本分组看，才知道问题出在哪一块。
  • 告警要少而准：每条告警都得有人处理，响得太多，大家很快就不看了。

收集用户反馈 ​

回答下面放 👍 / 👎，点击时把 trace id 一起发回后端：

python
@app.post("/api/feedback")          # app 是 A8 里的 FastAPI 应用
def feedback(body: dict):           # {"trace_id": "...", "good": false, "comment": "价格答错了"}
    write_span({"trace_id": body["trace_id"], "span": "feedback", "good": bool(body["good"]),
                "comment": redact(str(body.get("comment", "")))[:500]})
    return {"ok": True}

每周把 👎 的 trace 拿出来看：是检索没找到、模型理解错，还是工具失败？挑典型的写上正确答案，加进 H2 的评估集。线上出过的错，就成了以后每次改动都要通过的测试。👎 的比例也放进看板，按版本对比。

缓存 ​

精确缓存：同样的问题直接返回 ​

python
import hashlib

CACHE = {}                    # 演示用字典；正式环境用 Redis：SET key value EX ttl
PROMPT_VERSION = "faq-v2"

def cached_answer(trace_id, messages, model="gpt-5.5", ttl=3600):
    raw = json.dumps([model, PROMPT_VERSION, messages], ensure_ascii=False, sort_keys=True)
    key = hashlib.sha256(raw.encode()).hexdigest()
    hit = CACHE.get(key)
    if hit and hit["expires"] > time.time():
        write_span({"trace_id": trace_id, "span": "cache", "hit": True})
        return hit["text"]
    resp = traced_chat(trace_id, "answer", model=model, messages=messages, max_completion_tokens=500)
    text = resp.choices[0].message.content
    CACHE[key] = {"text": text, "expires": time.time() + ttl}
    return text
  • key 里包含模型、提示词版本和完整消息：改了提示词，旧缓存自然不再命中。用户问题先去掉首尾空格再算 key。
  • 一定要设 TTL：资料会更新，价格、活动这类内容缓存时间要短。
  • 只缓存和用户无关的回答（FAQ、产品说明）。带个人数据、依赖当前时间的回答不要缓存，更不能把 A 用户的回答返回给 B。
  • 看命中率：命中率很低，说明问题很分散，精确缓存帮不上忙，就别加这层复杂度。

提示词缓存：让上游复用相同的开头 ​

不少上游会缓存「和之前请求开头完全相同」的部分，命中的输入 token 更便宜、更快。能不能命中，取决于消息怎么排：

python
messages = [
    {"role": "system", "content": SYSTEM_PROMPT + "\n\n" + PRODUCT_DOCS},   # 长且固定：放在开头，一个字都别变
    *history[-6:],                                                           # 之前的对话，只往后追加
    {"role": "user", "content": question},                                   # 每次都变：放最后
]
resp = traced_chat(trace_id, "answer", model="gpt-5.5", messages=messages)
# span 里 cached_tokens > 0 说明开头命中了缓存；上游没有报告时这个字段为空，按未命中处理
  • 长而固定的放前面：system 提示词、工具定义、固定资料。
  • 会变的放最后：用户问题、检索结果、当前时间。
  • 开头不要放每次都变的东西（时间戳、随机编号、用户名）：开头差一个字，后面就都复用不了。

模型路由：简单的给便宜模型 ​

Yui把简单请求分给快模型，难题转给强模型
Yui把简单请求分给快模型，难题转给强模型AI 生成配图

大部分请求其实很简单。简单的交给便宜、快的模型，难的交给强模型，成本和延迟都能降下来。怎么分：

  1. 先用规则：很短的寒暄、命中 FAQ 关键词的，直接算简单；规则不花钱。
  2. 规则判断不了，再用便宜模型分类，用结构化输出返回档位和建议的输出上限。
  3. 拿不准就选强模型：把难题交给弱模型答错，代价通常比多花一点钱大。

试试这个分类器，换几种请求看它怎么分：

▶ 动手试试
系统提示词（这次请求一起发送，点开查看）
你是请求分类器，只判断请求难度，不回答问题。simple：寒暄、单个事实查询、短文本改写或翻译、格式转换，一步就能答。complex：需要多步推理、比较多个方案、计算、写长文或代码、结合用户自己的数据。拿不准时选 complex。max_tokens 是回答所需输出长度的估计：simple 一般 100–300，complex 一般 500–1500。needs_retrieval 表示回答是否依赖产品文档、订单记录等内部资料。
登录后运行登录 HiveGPT 后每天有免费运行次数
示例输出（之前运行的结果）
{
  "tier": "complex",
  "reason": "要比较三个套餐的退款规则，并结合用户的订单记录计算金额。",
  "max_tokens": 1000,
  "needs_retrieval": true
}

再试试「你好」「营业时间是几点？」「把这句话翻译成英文：明天见」，看档位和 max_tokens 怎么变。

接到代码里，分类失败或便宜模型出错时都要有退路：

python
from openai import APIConnectionError, APIStatusError, APITimeoutError

CHEAP_MODEL = "gpt-5.4-mini"   # 占位：从 HiveGPT 模型列表里选一个便宜、快的模型
STRONG_MODEL = "gpt-5.5"
ROUTER_SYSTEM = "你是请求分类器，只判断请求难度……"   # 同上面运行框里的 system
ROUTE_SCHEMA = {...}                                  # 同上面运行框里的 schema

def route(trace_id, question):
    if len(question) <= 10 and not any(w in question for w in ("代码", "比较", "计算", "方案")):
        return {"tier": "simple", "max_tokens": 300, "needs_retrieval": False}   # 规则：短问题
    resp = traced_chat(trace_id, "route", model=CHEAP_MODEL, max_completion_tokens=300,
                       messages=[{"role": "system", "content": ROUTER_SYSTEM}, {"role": "user", "content": question}],
                       response_format={"type": "json_schema", "json_schema": ROUTE_SCHEMA})
    return json.loads(resp.choices[0].message.content)

def answer(trace_id, question):
    try:
        r = route(trace_id, question)
    except Exception:                                    # 分类失败：按难题处理
        r = {"tier": "complex", "max_tokens": 1500, "needs_retrieval": True}
    model = STRONG_MODEL if r["tier"] == "complex" else CHEAP_MODEL
    max_tokens = min(max(r["max_tokens"], 100), 2000)    # 模型给的数字也要夹在合理范围内
    messages = [{"role": "system", "content": "你是简洁的客服助手。"}, {"role": "user", "content": question}]
    # needs_retrieval 为 True 时，先按 H4 检索资料再放进 messages，这里省略
    try:
        return traced_chat(trace_id, "answer", model=model, messages=messages, max_completion_tokens=max_tokens)
    except (APIStatusError, APIConnectionError, APITimeoutError):
        if model == STRONG_MODEL:
            raise
        return traced_chat(trace_id, "answer-fallback", model=STRONG_MODEL,
                           messages=messages, max_completion_tokens=max_tokens)

上线后按档位看 👎 比例和费用。如果 simple 档的 👎 明显偏多，说明分得太激进，调规则或分类提示词。分类器本身也可以用 H2 的方法建一个小评估集来测。

成本控制 ​

A8 的方法（控制上下文、限制输出、按任务选模型）之外，生产中还常用：

  • 按档位设 max_tokens：分类只要几十到几百个 token，简单问答几百，长文再放开。
  • 缩短上下文：历史对话做摘要（H3），检索只带相关度高的几段（H4），工具结果先截断再交回。
  • 离线任务批量做：夜间汇总、批量打标签这类不需要实时的任务，把多条短文本合进一次请求，用便宜模型，控制并发慢慢跑。
  • 每个用户设额度：除了 A8 的请求次数限流，再按 token 或费用设每日上限，超了就降级到便宜模型或提示明天再来。
  • 按功能算账：从 trace 里按功能汇总费用，先优化开销大的功能。

安全上线：版本、灰度、回滚 ​

提示词和模型都写进配置并标上版本号，每个 span 记下当时用的版本：

python
import hashlib

RELEASE = {"current": {"prompt": "support-v3", "model": "gpt-5.5"},
           "canary":  {"prompt": "support-v4", "model": "gpt-5.5"},
           "canary_percent": 5}

def pick_release(user_id):
    bucket = int(hashlib.md5(user_id.encode()).hexdigest(), 16) % 100   # 同一用户总落在同一组
    return RELEASE["canary"] if bucket < RELEASE["canary_percent"] else RELEASE["current"]
  1. 先跑评估：用 H2 的评估集跑新版本，通过率不低于旧版本才继续。
  2. 小比例灰度：先给 5% 的用户，看一两天的错误率、p95、费用和 👎 比例，和旧版本对比。
  3. 逐步放大：没问题再加到 20%、50%、100%。
  4. 随时回滚：有问题把 canary_percent 改回 0，不用重新部署。

一次只改一样东西。同时换模型和提示词，变好变坏都说不清是谁的原因。

结业项目：一个能上线的 AI 应用 ​

做一个真实场景的 AI 应用，例如客服助手（回答某个产品或店铺的问题）或学习助手（按教材答疑、出练习题），可以在 A8 的聊天机器人上继续做。它要包含：

  1. 评估集（H2）：建议至少 20 条用例和一个运行脚本，README 里写上最近一次的通过率。
  2. 护栏（H6）：至少一项，例如提示注入检查、输出里的敏感信息过滤，或有副作用操作的人工确认（H7）。
  3. 日志和追踪：每个请求一个 trace id，记录模型、token、耗时、费用和错误；提示词脱敏后再存。
  4. 一项成本优化：缓存、模型路由、缩短上下文任选一项，在 README 里写出优化前后的 token 或费用对比。

提交方式任选其一：

  • 用 HiveGPT「我的网站」（网站托管）发布项目页面：介绍功能，放截图、评估结果和成本对比。它托管的是静态网页，后端和 Key 要放在你自己的服务器上。
  • 提交公开的 GitHub 或 Gitee 仓库链接。仓库里不要有 Key，用 .gitignore 排除 .env。

小结 ​

  • 每个请求一个 trace，每次调用一条 span，记模型、token、耗时、费用、错误；提示词脱敏后再存；👎 连同 trace id 进评估集。
  • 精确缓存处理和用户无关的重复问题，提示词缓存靠「固定的开头放前面」；模型路由让简单请求走便宜模型，出错回退到强模型。
  • 提示词和模型都带版本，评估通过后再灰度上线，有问题改配置就能回滚。

这是 H 路线的最后一课。完成全部测验和检查点、提交结业项目后，就可以到路线页领取结业证书。

代码示例在页面里运行时使用 HiveGPT 的模型接口。延伸阅读来自 JavaGuide（Apache-2.0），版权归原作者。

© HiveGPT