让大模型说"穴居人语言"，Token 直接省 65%？这个火出圈的玩法我替你踩过坑了

import time
import json
import requests

# ========== Token 统计工具 ==========
class TokenCounter:
    """简易 Token 统计：中文按 1.5 字/Token 估算，英文按 0.75 词/Token"""

    @staticmethod
    def estimate(text: str) -> int:
        cn_chars = sum(1 for ch in text if '\u4e00' <= ch <= '\u9fff')
        other_chars = len(text) - cn_chars
        # 中文约 1.5 字一个 token，其他字符约 4 字符一个 token
        return int(cn_chars / 1.5 + other_chars / 4)


# ========== 两套 Prompt 模板 ==========
NORMAL_SYSTEM_PROMPT = """你是一位资深的技术专家。请详细、专业地回答用户的问题，
包含背景说明、分析过程、具体步骤和注意事项。回答要完整清晰。"""

CAVEMAN_SYSTEM_PROMPT = """You must respond in caveman speak.
短句。只留关键动作和结论。不用敬语，不要解释原理，不要"首先其次"。
像原始人说话：火好。肉香。冰冷。
技术内容不许错，但表达要最短。"""

# 定价（示例：某国产模型，元/百万Token）
PRICE_INPUT = 2.0    # 输入单价
PRICE_OUTPUT = 8.0   # 输出单价（一般是输入的4倍）


def run_compare(question: str):
    """同一问题跑两种模式，对比 Token 和耗时"""
    results = {}

    for mode, sys_prompt in [("normal", NORMAL_SYSTEM_PROMPT),
                              ("caveman", CAVEMAN_SYSTEM_PROMPT)]:
        start = time.time()
        response = requests.post(
            "http://localhost:8000/v1/chat/completions",
            json={
                "model": "qwen2.5-14b-instruct",
                "messages": [
                    {"role": "system", "content": sys_prompt},
                    {"role": "user", "content": question}
                ],
                "temperature": 0.3,
            },
            timeout=60
        )
        latency = time.time() - start
        answer = response.json()["choices"][0]["message"]["content"]

        in_tokens = TokenCounter.estimate(sys_prompt + question)
        out_tokens = TokenCounter.estimate(answer)

        results[mode] = {
            "answer": answer,
            "in_tokens": in_tokens,
            "out_tokens": out_tokens,
            "latency": round(latency, 2),
            "cost": round((in_tokens * PRICE_INPUT +
                           out_tokens * PRICE_OUTPUT) / 1_000_000, 6)
        }

    return results


# ========== 批量测试 ==========
test_questions = [
    "服务器 CPU 使用率过高怎么排查？",
    "数据库连接池满了会出现什么现象？",
    "如何设计一个分布式锁？",
]

print(f"{'问题':<20} {'模式':<10} {'输入Tok':<10} {'输出Tok':<10} "
      f"{'耗时(s)':<10} {'成本(元)':<12}")
print("-" * 80)

total_normal_cost = 0
total_caveman_cost = 0

for q in test_questions:
    r = run_compare(q)
    for mode in ["normal", "caveman"]:
        d = r[mode]
        print(f"{q[:18]:<20} {mode:<10} {d['in_tokens']:<10} "
              f"{d['out_tokens']:<10} {d['latency']:<10} {d['cost']:<12}")
        if mode == "normal":
            total_normal_cost += d["cost"]
        else:
            total_caveman_cost += d["cost"]

print("-" * 80)
saving = (1 - total_caveman_cost / total_normal_cost) * 100
print(f"普通模式总成本: {total_normal_cost:.6f} 元")
print(f"穴居人模式总成本: {total_caveman_cost:.6f} 元")
print(f"节省: {saving:.1f}%")

# ===== 实测输出示例 =====
# 问题                    模式        输入Tok     输出Tok     耗时(s)     成本(元)
# --------------------------------------------------------------------------------
# 服务器CPU过高怎么排查？  normal      45          380         4.2         0.00313
# 服务器CPU过高怎么排查？  caveman     78          65          0.9         0.00068
# 数据库连接池满现象？      normal      42          350         3.8         0.00288
# 数据库连接池满现象？      caveman     75          58          0.8         0.00061
# 如何设计分布式锁？        normal      38          520         5.5         0.00430
# 如何设计分布式锁？        caveman     72          85          1.1         0.00082
# --------------------------------------------------------------------------------
# 普通模式总成本: 0.010310 元
# 穴居人模式总成本: 0.002110 元
# 节省: 79.5%

# 穴居人模式实际输出示例（问：服务器 CPU 过高怎么排查）：
# "先看 top。哪个进程吃 CPU。Java 进程大？看线程。jstack 抓。
#  系统进程大？看日志。io 等待高？查磁盘。
#  设告警。CPU 80% 报。别等死机才看。"

import json
import time
from datetime import datetime
import requests

class OpsDailyReportGenerator:
    """运维日报生成器（穴居人优化版）"""

    def __init__(self, api_url, model_name):
        self.api_url = api_url
        self.model = model_name

    def generate_log_summary(self, server, logs):
        """
        单台服务器的日志分析
        关键改造：系统提示词切换成穴居人模式 + 明确输出结构
        """
        system_prompt = """你是运维日志分析器。穴居人模式回复：
规则：
1. 每条问题最多两行：问题 + 动作
2. 不解释原理。不写"建议关注"。不写客套话
3. 危险等级打头：[红]=马上处理 [黄]=今天处理 [绿]=记录在案
4. 没问题的服务器只回一行：无事
5. 输出必须是 JSON 数组，字段：level, issue, action"""

        logs_text = "\n".join(logs[-50:])  # 最多取50条，防止超长

        response = requests.post(
            f"{self.api_url}/v1/chat/completions",
            json={
                "model": self.model,
                "messages": [
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content":
                        f"服务器: {server}\n异常日志:\n{logs_text}"}
                ],
                "temperature": 0.1,
                "max_tokens": 400,   # 穴居人模式下 400 绰绰有余
            },
            timeout=30
        )
        return response.json()["choices"][0]["message"]["content"]

    def generate_daily_report(self, all_server_logs):
        """汇总所有服务器，生成日报"""
        report_items = []
        total_in_tokens = 0
        total_out_tokens = 0

        for server, logs in all_server_logs.items():
            raw = self.generate_log_summary(server, logs)
            try:
                items = json.loads(raw)
                report_items.extend([
                    {"server": server, **item} for item in items
                ])
            except json.JSONDecodeError:
                # 兜底：解析失败就存原文，不丢数据
                report_items.append({
                    "server": server, "level": "[黄]",
                    "issue": "AI输出解析失败", "action": raw[:100]
                })

        # 按危险等级排序：红 -> 黄 -> 绿
        level_order = {"[红]": 0, "[黄]": 1, "[绿]": 2}
        report_items.sort(key=lambda x: level_order.get(x["level"], 3))

        return self._render_report(report_items)

    def _render_report(self, items):
        """渲染日报——因为模型输出已经是短句，这步几乎没有二次加工"""
        today = datetime.now().strftime("%Y-%m-%d")
        lines = [f"📋 运维日报 {today}", "=" * 40]

        current_level = None
        for item in items:
            if item["level"] != current_level:
                current_level = item["level"]
                lines.append(f"\n{current_level}")
            lines.append(
                f"  {item['server']} | {item['issue']} → {item['action']}"
            )

        return "\n".join(lines)


# ===== 实际运行效果 =====
# 改造前的日报（模型普通模式输出）：
# "服务器 web-prod-01 在过去24小时内出现了多次内存使用率告警，
#  经分析主要原因是 Java 应用存在内存缓慢增长的趋势，这可能与
#  存在内存泄漏相关。建议您尽快安排人员检查应用的堆内存快照，
#  并关注 Full GC 的频率变化情况……"（单台服务器约 300 字）
#
# 改造后的日报（穴居人模式输出）：
# "[红] web-prod-01 | 内存连涨6小时，Full GC频繁 → 抓heap dump，今天查泄漏
#  [黄] db-slave-02 | 主从延迟30s → 查大事务，必要时重启同步
#  [绿] cache-01 | 昨晚重启后正常 → 记录在案"
#
# 同样的信息，字数从 300 降到 40，工程师还更好读了

generator = OpsDailyReportGenerator(
    api_url="http://localhost:8000",
    model_name="qwen2.5-14b-instruct"
)