1. 项目概述：从“烧钱”到“省钱”的智能诊断之旅

如果你也用过 Claude 这类大型语言模型 API，大概率经历过这种场景：项目进展顺利，代码写得飞快，但月底账单寄来时，心里“咯噔”一下——怎么又超预算了？Token 消耗就像个无底洞，尤其是当你把 Claude 集成到自动化流程、聊天机器人或者日常开发辅助工具里时，那些看似不起眼的交互，累积起来就是一笔可观的费用。更让人头疼的是，你往往不清楚这些 Token 到底“烧”在了哪里。是提示词写得过于冗长？还是模型返回了太多无关内容？抑或是你的代码在循环调用时产生了不必要的重复请求？

这正是我过去几个月里反复经历的困境。作为一个深度依赖 Claude API 进行内容生成、代码审查和数据分析的开发者，我发现自己陷入了“开发-测试-收到天价账单-恐慌-试图优化-效果不明-继续开发”的恶性循环。直到有一次，一个本应简单的数据清洗脚本，因为一个循环逻辑的疏忽，在夜间无人值守时连续调用了数百次 API，直接导致当月成本飙升了 300%。那一刻我意识到，不能再靠“感觉”和“手动检查”来管理 Token 消耗了。我需要一个能透视整个调用过程的“诊断工具”，它能精准地告诉我：每一分钱花在了哪里，为什么花，以及如何更聪明地花。

于是，“CLAUDE.md 诊断工具”这个项目诞生了。它的核心目标不是简单地记录 Token 数量，而是像给 API 调用流程做一次全面的“体检”。它要能解析每一次请求和响应的结构，量化提示词中各个组成部分的消耗，识别低效或冗余的模式，并最终给出具体、可操作的优化建议。这个工具的本质，是将 API 使用的“黑盒”变成“白盒”，让成本控制从一种被动的、事后的补救，转变为主动的、基于数据的工程实践。接下来，我将详细拆解我是如何一步步构建这个工具的，从核心思路到技术选型，从关键实现到避坑经验，希望能为同样受困于 Token 成本的朋友们提供一条清晰的解决路径。

2. 核心思路与架构设计：构建一个“Token 审计员”

2.1 从痛点出发：我们需要诊断什么？

在动手写代码之前，我花了大量时间梳理 Claude API 调用中可能导致 Token 消耗失控的典型场景。这决定了工具需要具备哪些诊断能力。我将其归纳为四个核心维度：

1. 提示词（Prompt）结构分析 ：这是最大的成本变量。一个糟糕的提示词可能包含大量冗余的上下文、重复的指令或不必要的示例。工具需要能解析提示词，区分系统指令、用户消息、历史对话、示例等部分，并量化每一部分的 Token 占比。例如，一个用于代码生成的提示词里，如果包含了长达 100 行的无关历史代码作为上下文，那么其成本效率必然低下。

2. 请求/响应模式审计 ：很多消耗源于非优化的调用模式。例如，频繁发送内容相似的短请求（每次调用都携带完整的系统指令），而不是利用好对话历史；或者在循环中重复请求相同或微调的内容。工具需要能追踪请求序列，识别出这类模式。

3. 内容效率评估 ：模型返回的内容是否“物有所值”？有时，我们请求一个摘要，模型却返回了包含大量分析过程的冗长文本；有时，我们只需要一个“是/否”的判断，模型却生成了一段解释。工具需要能结合请求的意图（通过提示词推断）和响应的内容，评估其信息密度和必要性。

4. 配置与参数审查 ：API 调用时的参数设置直接影响成本和效果。例如， max_tokens 参数设置得过高，模型可能会生成远超需要的内容； temperature 设置不当可能导致生成内容不稳定，需要多次重试。工具需要检查这些参数设置的合理性。

基于以上分析，我决定将工具设计成一个轻量级的中间件（Middleware）或装饰器（Decorator），它能够无缝集成到现有的代码库中，拦截所有对 Claude API 的调用，收集详尽的元数据，并进行离线或准实时分析。

2.2 技术栈选型与架构决策

为了平衡灵活性、性能和易用性，我选择了以下技术栈：

• 核心语言：Python 。这是与 Claude API 交互最广泛的生态语言，拥有成熟的 HTTP 客户端库（如 httpx , aiohttp ）和丰富的异步支持，便于处理高并发下的日志收集。
• 数据收集层：自定义 HTTP 适配器与装饰器 。我放弃了直接修改业务代码，而是采用侵入性更低的方式。对于使用 requests 或 httpx 库的同步/异步客户端，我通过实现自定义的 HTTPAdapter 或事件钩子（Hooks）来拦截请求和响应。对于使用 Anthropic 官方 SDK 或其他封装库的情况，我编写了装饰器来包裹关键的调用函数。这样，业务代码几乎无需改动。
• 元数据存储：结构化日志与本地 SQLite 。为了便于后续分析，我将每次调用的元数据（时间戳、请求体、响应头、原始响应、计算出的 Token 数、自定义标签等）以结构化的 JSON 格式写入日志文件。同时，为了支持快速的交互式查询和聚合分析，我设计了一个轻量级的数据管道，定期（或按需）将这些日志解析并存入一个本地 SQLite 数据库。SQLite 足够轻量，无需额外服务，且通过索引可以高效地进行复杂查询。
• 分析引擎：Pandas + 启发式规则 。数据分析部分使用 Pandas 从 SQLite 中加载数据，进行聚合、分组和计算。诊断逻辑则基于一系列我总结的启发式规则（Heuristic Rules）来实现。例如，一条规则是：“如果连续 5 次请求的系统指令（System Prompt）完全相同，且用户消息（User Message）内容相似度超过 80%，则标记为‘可能可合并的重复指令’。”
• 报告生成：Jinja2 模板 + Markdown/HTML 。诊断结果需要以清晰易懂的方式呈现。我使用 Jinja2 模板引擎来生成报告，支持输出为 Markdown 文件（便于在版本控制中跟踪）和 HTML 文件（便于可视化阅读）。报告会包含概览仪表板、Top N 消耗请求详情、具体的优化建议列表等。

整个架构的流程图如下（概念描述）：

[你的应用代码] -> [诊断工具装饰器/中间件] -> [Claude API]
                                    |
                                    v
                            [结构化日志文件]
                                    |
                                    v (解析入库)
                            [SQLite 数据库]
                                    |
                                    v (分析引擎)
                            [Pandas DataFrame]
                                    |
                                    v (规则应用)
                            [诊断结果与建议]
                                    |
                                    v (报告渲染)
                            [Markdown/HTML 报告]

这个架构的关键优势在于 非侵入性 和 灵活性 。开发者只需在初始化 API 客户端时加入几行配置代码，即可开启全量审计。所有复杂的分析都在后台异步完成，不影响主业务流程的性能。

3. 核心模块实现细节

3.1 请求/响应拦截与元数据捕获

这是工具的数据源头，必须保证捕获的信息既全面又准确。我以 httpx 异步客户端为例，展示核心实现思路。

首先，我创建了一个自定义的传输层（Transport），它继承自默认的传输类，并重写了 handle_async_request 方法。在这个方法中，我复制了请求的副本，然后发起真正的网络请求。在收到响应后，我不仅返回响应给上游，还异步地将请求和响应的关键信息发送到一个处理队列中。

import httpx
import asyncio
import json
from datetime import datetime
from typing import Dict, Any
import hashlib

class ClaudeDiagnosticsTransport(httpx.AsyncHTTPTransport):
    """自定义传输层，用于拦截 Claude API 调用"""
    def __init__(self, log_queue: asyncio.Queue, **kwargs):
        super().__init__(**kwargs)
        self.log_queue = log_queue

    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
        # 1. 捕获请求信息
        request_body = await request.aread()
        request_data = {
            "url": str(request.url),
            "method": request.method,
            "headers": dict(request.headers),
            "body": request_body.decode('utf-8') if request_body else None,
            "timestamp": datetime.utcnow().isoformat(),
        }
        # 给请求生成一个唯一ID，便于关联
        request_hash = hashlib.md5(f"{request_data['url']}{request_data['timestamp']}".encode()).hexdigest()
        request_data['request_id'] = request_hash

        # 2. 重新构建请求体，因为 aread() 消耗了它
        request.stream = httpx.ByteStream(request_body)

        # 3. 发送原始请求
        response = await super().handle_async_request(request)

        # 4. 捕获响应信息
        response_body = await response.aread()
        response_data = {
            "request_id": request_hash,
            "status_code": response.status_code,
            "headers": dict(response.headers),
            "body": response_body.decode('utf-8') if response_body else None,
            "timestamp": datetime.utcnow().isoformat(),
        }
        # 重新构建响应体
        response.stream = httpx.ByteStream(response_body)

        # 5. 将元数据放入队列，供后台任务处理
        log_entry = {
            "type": "api_call",
            "request": request_data,
            "response": response_data
        }
        await self.log_queue.put(log_entry)

        return response

关键细节与注意事项：

• 请求体消耗问题 ： aread() 方法会消耗请求体的流，所以必须在调用后重新构建 request.stream ，否则后续真正的网络请求会失败。响应体同理。
• 异步非阻塞 ：日志入队操作是异步的，且使用了队列，确保了拦截操作不会阻塞主请求-响应链路。后台有一个独立的任务从队列中取出数据，进行 Token 计算和写入磁盘。
• 敏感信息处理 ：在真实环境中，API 密钥（通常存在于 Authorization 头中）必须被脱敏后再记录。我通常在日志处理环节将其替换为 *** 。
• Token 计算 ：Claude API 的响应头中通常包含 x-amzn-bedrock-input-token-count 和 x-amzn-bedrock-output-token-count （如果使用 Amazon Bedrock）或自定义的类似头字段。但为了更通用和精确，我实现了一个本地的 Token 计数器，使用与 Claude 模型相同的分词器（例如 tiktoken ，虽然 Claude 有自己的分词方式，但 tiktoken 的 cl100k_base 编码对于估算非常接近）。这样即使响应头没有提供信息，我们也能计算。

3.2 Token 计算与成本估算模块

准确的 Token 计数是诊断的基础。我创建了一个 TokenAnalyzer 类。

import tiktoken
from anthropic import Anthropic

class TokenAnalyzer:
    def __init__(self):
        # 使用 tiktoken 进行近似估算（速度快，适用于本地分析）
        self._tiktoken_encoder = tiktoken.get_encoding("cl100k_base")
        # 对于更精确的场景，可以初始化 Anthropic 客户端（需要 API Key）
        self._anthropic_client = None # Anthropic(api_key="...")

    def estimate_tokens_tiktoken(self, text: str) -> int:
        """使用 tiktoken 快速估算 Token 数。对于中文混合文本，这是一个较好的近似值。"""
        if not text:
            return 0
        return len(self._tiktoken_encoder.encode(text))

    def count_tokens_anthropic(self, messages: list) -> int:
        """使用 Anthropic 官方 SDK 进行精确计数（会产生一次额外的 API 调用，慎用）。"""
        if not self._anthropic_client:
            # 回退到估算模式
            # 将 messages 列表拼接成字符串进行估算（不精确）
            combined_text = " ".join([f"{m['role']}: {m['content']}" for m in messages])
            return self.estimate_tokens_tiktoken(combined_text)
        # 注意：官方 SDK 的 count_tokens 方法可能随版本变化
        try:
            return self._anthropic_client.count_tokens(messages)
        except:
            # 如果失败，回退到估算
            combined_text = " ".join([f"{m['role']}: {m['content']}" for m in messages])
            return self.estimate_tokens_tiktoken(combined_text)

    def analyze_prompt_structure(self, prompt_data: Dict[str, Any]) -> Dict[str, Any]:
        """解析提示词结构，并估算各部分的 Token 消耗。"""
        # prompt_data 可能是字符串，也可能是 OpenAI/Anthropic 格式的消息列表
        analysis = {"total_tokens": 0, "parts": []}
        if isinstance(prompt_data, str):
            # 简单字符串提示词
            tokens = self.estimate_tokens_tiktoken(prompt_data)
            analysis["total_tokens"] = tokens
            analysis["parts"].append({"role": "user", "content_preview": prompt_data[:100], "token_count": tokens})
        elif isinstance(prompt_data, list):
            # 消息列表格式，例如 [{"role": "user", "content": "..."}, {"role": "assistant", "content": "..."}]
            for msg in prompt_data:
                role = msg.get("role", "unknown")
                content = msg.get("content", "")
                tokens = self.estimate_tokens_tiktoken(str(content))
                analysis["total_tokens"] += tokens
                analysis["parts"].append({
                    "role": role,
                    "content_preview": str(content)[:100],
                    "token_count": tokens,
                    "percentage": 0 # 稍后计算
                })
            # 计算百分比
            for part in analysis["parts"]:
                if analysis["total_tokens"] > 0:
                    part["percentage"] = round((part["token_count"] / analysis["total_tokens"]) * 100, 2)
        return analysis

    def calculate_cost(self, input_tokens: int, output_tokens: int, model: str = "claude-3-sonnet-20240229") -> float:
        """根据模型和 Token 数计算估算成本（美元）。价格需要定期更新。"""
        # 示例价格表（单位：美元/每千Token），请根据 Anthropic 最新定价更新
        price_map = {
            "claude-3-5-sonnet-20241022": {"input": 0.003, "output": 0.015},
            "claude-3-opus-20240229": {"input": 0.015, "output": 0.075},
            "claude-3-sonnet-20240229": {"input": 0.003, "output": 0.015},
            "claude-3-haiku-20240307": {"input": 0.00025, "output": 0.00125},
        }
        model_prices = price_map.get(model, price_map["claude-3-sonnet-20240229"])
        cost = (input_tokens / 1000) * model_prices["input"] + (output_tokens / 1000) * model_prices["output"]
        return round(cost, 4)

实操心得：

• 精度与效率的权衡 ： tiktoken 估算速度极快，对于诊断和趋势分析完全够用。只有在需要极其精确的、按次计费的场景下，才值得调用官方的 count_tokens 方法（它本身也可能消耗资源）。我的工具默认使用 tiktoken 估算，并在报告中注明这是“估算值”。
• 成本映射 ：模型价格会变动，我将价格表设计为可配置的字典，并提醒用户定期从 Anthropic 官网更新。在报告中，我会同时显示 Token 数量和估算成本，让开发者对“钱花在哪”有更直观的感受。
• 结构化分析 ： analyze_prompt_structure 函数是诊断的“眼睛”。它能清晰地展示一个复杂提示词中，系统指令、用户问题、历史对话、示例各自占了多少“篇幅”。很多时候，优化就从这里开始——比如发现系统指令占了总 Token 的 40%，那么就可以考虑是否过于冗长，或者能否将其精简并固化。

3.3 诊断规则引擎的实现

诊断的核心是一系列规则。每条规则都是一个独立的函数，接收一次 API 调用记录或一组记录，返回一个诊断结果对象。我使用 Python 的 dataclass 来定义诊断结果。

from dataclasses import dataclass
from typing import List, Optional
import re

@dataclass
class Diagnosis:
    rule_id: str          # 规则唯一标识，如 "RULE_001"
    severity: str         # 严重程度: HIGH, MEDIUM, LOW
    title: str            # 问题标题
    description: str      # 详细描述
    suggestion: str       # 具体优化建议
    affected_request_ids: List[str]  # 关联的请求ID
    context: Optional[Dict] = None  # 额外的上下文数据，如 Token 数、重复内容等

class RuleEngine:
    def __init__(self):
        self.rules = []
        self._register_rules()

    def _register_rules(self):
        """注册所有诊断规则"""
        self.rules.append(self._rule_redundant_system_prompt)
        self.rules.append(self._rule_high_output_to_input_ratio)
        self.rules.append(self._rule_similar_consecutive_requests)
        # ... 可以添加更多规则

    def run_diagnosis(self, call_logs: List[Dict]) -> List[Diagnosis]:
        """对一组调用日志运行所有诊断规则"""
        all_diagnoses = []
        for rule_func in self.rules:
            diagnoses = rule_func(call_logs)
            if diagnoses:
                all_diagnoses.extend(diagnoses)
        return all_diagnoses

    def _rule_redundant_system_prompt(self, call_logs: List[Dict]) -> List[Diagnosis]:
        """规则1：检测冗长的系统提示词"""
        diagnoses = []
        for log in call_logs:
            request_body = log.get('request', {}).get('body')
            if not request_body:
                continue
            try:
                data = json.loads(request_body)
                # 假设请求体格式包含 `messages` 或 `prompt`
                messages = data.get('messages', [])
                system_prompt = None
                for msg in messages:
                    if msg.get('role') == 'system':
                        system_prompt = msg.get('content')
                        break
                if system_prompt:
                    token_analyzer = TokenAnalyzer()
                    sys_tokens = token_analyzer.estimate_tokens_tiktoken(system_prompt)
                    # 如果系统提示词超过 500 tokens，标记为可能冗长
                    if sys_tokens > 500:
                        diagnoses.append(Diagnosis(
                            rule_id="RULE_001",
                            severity="MEDIUM",
                            title="冗长的系统提示词",
                            description=f"系统指令消耗了 {sys_tokens} 个 Token，可能包含过多一次性上下文或过于详细的约束。",
                            suggestion="考虑将静态的、冗长的上下文信息移出系统提示词，改用文件上传（File Upload）功能，或在首次对话中提供，后续通过 `{context}` 占位符引用。",
                            affected_request_ids=[log.get('request', {}).get('request_id')],
                            context={"system_prompt_preview": system_prompt[:200], "token_count": sys_tokens}
                        ))
            except json.JSONDecodeError:
                continue
        return diagnoses

    def _rule_high_output_to_input_ratio(self, call_logs: List[Dict]) -> List[Diagnosis]:
        """规则2：检测输出/输入 Token 比例过高的情况"""
        diagnoses = []
        for log in call_logs:
            # 从日志中提取估算的输入输出 Token 数（在日志处理阶段已计算并存入）
            input_tokens = log.get('analysis', {}).get('input_tokens_est', 0)
            output_tokens = log.get('analysis', {}).get('output_tokens_est', 0)
            if input_tokens > 0 and output_tokens > 0:
                ratio = output_tokens / input_tokens
                # 如果输出是输入的 5 倍以上，且输出绝对量较大，可能意味着提示词引导不足或 max_tokens 设置过大
                if ratio > 5.0 and output_tokens > 1000:
                    diagnoses.append(Diagnosis(
                        rule_id="RULE_002",
                        severity="LOW", # 可能是预期行为，所以严重程度较低
                        title="高输出/输入 Token 比例",
                        description=f"本次调用输出 Token 数是输入 Token 数的 {ratio:.1f} 倍（输入：{input_tokens}，输出：{output_tokens}）。",
                        suggestion="检查提示词是否足够明确以引导模型生成简洁回答。考虑适当降低 `max_tokens` 参数，或使用‘请用一句话回答’、‘请列出要点’等指令来约束输出长度。",
                        affected_request_ids=[log.get('request', {}).get('request_id')],
                        context={"input_tokens": input_tokens, "output_tokens": output_tokens, "ratio": ratio}
                    ))
        return diagnoses

    def _rule_similar_consecutive_requests(self, call_logs: List[Dict]) -> List[Diagnosis]:
        """规则3：检测内容高度相似的连续请求（可能意味着循环逻辑浪费）"""
        if len(call_logs) < 2:
            return []
        diagnoses = []
        # 按时间排序
        sorted_logs = sorted(call_logs, key=lambda x: x.get('request', {}).get('timestamp', ''))
        for i in range(len(sorted_logs) - 1):
            req1 = sorted_logs[i].get('request', {}).get('body', '')
            req2 = sorted_logs[i+1].get('request', {}).get('body', '')
            # 简单的相似度检查：计算 Jaccard 相似度（基于单词集合）
            # 在实际应用中，可以使用更复杂的文本相似度算法，如 TF-IDF + 余弦相似度
            words1 = set(re.findall(r'\w+', req1.lower()))
            words2 = set(re.findall(r'\w+', req2.lower()))
            if words1 and words2:
                intersection = len(words1.intersection(words2))
                union = len(words1.union(words2))
                similarity = intersection / union if union > 0 else 0
                # 如果相似度超过 0.7，且时间间隔很短（例如小于10秒），则标记
                time1 = sorted_logs[i].get('request', {}).get('timestamp')
                time2 = sorted_logs[i+1].get('request', {}).get('timestamp')
                # 此处省略时间间隔计算逻辑...
                if similarity > 0.7:
                    diagnoses.append(Diagnosis(
                        rule_id="RULE_003",
                        severity="HIGH",
                        title="检测到高度相似的连续请求",
                        description=f"相邻两次请求的内容相似度约为 {similarity:.2%}，可能由于循环、重试逻辑或用户快速重复提交导致。",
                        suggestion="检查代码逻辑，避免在循环中发送相同或微调的请求。考虑实现请求去重缓存（对相同输入缓存响应），或增加用户操作的防抖（Debounce）机制。",
                        affected_request_ids=[
                            sorted_logs[i].get('request', {}).get('request_id'),
                            sorted_logs[i+1].get('request', {}).get('request_id')
                        ],
                        context={"similarity": similarity}
                    ))
        return diagnoses

规则设计经验：

• 严重程度分级 ：我将规则分为高、中、低三个等级。 HIGH 通常指直接导致资源浪费的错误（如循环重复请求）， MEDIUM 指可优化的低效模式（如冗长提示词）， LOW 则更多是提示性建议（如输出比例高，但这在某些场景下是合理的）。
• 避免误报 ：规则的条件设置需要一定的阈值和上下文判断。例如， _rule_similar_consecutive_requests 规则不仅看相似度，还应结合时间间隔。如果两次请求间隔一小时，即使内容相似，也可能是正常的独立操作。
• 规则可扩展 ：整个规则引擎设计为插件式，新的诊断规则只需编写一个函数并注册到 self.rules 列表中即可。这使得工具可以随着对 Claude API 使用模式的深入理解而不断进化。

3.4 数据聚合分析与报告生成

收集了日志、运行了诊断之后，最后一步是将洞察可视化。我使用 Pandas 进行数据聚合，并用 Jinja2 生成报告。

import pandas as pd
from jinja2 import Environment, FileSystemLoader
import sqlite3
from pathlib import Path

class ReportGenerator:
    def __init__(self, db_path: str):
        self.db_path = db_path
        self.env = Environment(loader=FileSystemLoader('templates/')) # 假设模板在 templates 目录

    def generate_report(self, start_time: str, end_time: str, output_path: str):
        """生成指定时间范围内的诊断报告"""
        # 1. 从 SQLite 数据库加载数据
        conn = sqlite3.connect(self.db_path)
        query = """
        SELECT request_id, timestamp, model, input_tokens_est, output_tokens_est, 
               estimated_cost, prompt_structure, diagnosis_results
        FROM api_calls 
        WHERE timestamp BETWEEN ? AND ?
        ORDER BY timestamp
        """
        df = pd.read_sql_query(query, conn, params=(start_time, end_time))
        conn.close()

        if df.empty:
            print("指定时间段内无数据。")
            return

        # 2. 计算聚合指标
        total_calls = len(df)
        total_input_tokens = df['input_tokens_est'].sum()
        total_output_tokens = df['output_tokens_est'].sum()
        total_cost = df['estimated_cost'].sum()
        avg_cost_per_call = total_cost / total_calls if total_calls > 0 else 0

        # 按模型分组统计
        model_stats = df.groupby('model').agg({
            'request_id': 'count',
            'input_tokens_est': 'sum',
            'output_tokens_est': 'sum',
            'estimated_cost': 'sum'
        }).rename(columns={'request_id': 'call_count'}).reset_index()

        # 3. 提取诊断结果
        all_diagnoses = []
        for _, row in df.iterrows():
            if row['diagnosis_results']:
                try:
                    diagnoses = json.loads(row['diagnosis_results'])
                    all_diagnoses.extend(diagnoses)
                except:
                    pass

        # 4. 使用 Jinja2 渲染报告
        template = self.env.get_template('diagnosis_report.md.j2')
        report_content = template.render(
            period_start=start_time,
            period_end=end_time,
            total_calls=total_calls,
            total_input_tokens=int(total_input_tokens),
            total_output_tokens=int(total_output_tokens),
            total_cost=round(total_cost, 2),
            avg_cost_per_call=round(avg_cost_per_call, 4),
            model_stats=model_stats.to_dict('records'),
            diagnoses=all_diagnoses,
            # 可以传入更多数据，如消耗最高的 Top 10 请求等
            top_expensive_requests=df.nlargest(10, 'estimated_cost').to_dict('records')
        )

        # 5. 写入文件
        Path(output_path).parent.mkdir(parents=True, exist_ok=True)
        with open(output_path, 'w', encoding='utf-8') as f:
            f.write(report_content)
        print(f"报告已生成: {output_path}")

对应的 Jinja2 模板 ( templates/diagnosis_report.md.j2 ) 示例：

# Claude API 使用诊断报告
**分析时段**: {{ period_start }} 至 {{ period_end }}

## 概览
- **总调用次数**: {{ total_calls }}
- **总输入 Token**: {{ total_input_tokens | default(0) | int | format_number }}
- **总输出 Token**: {{ total_output_tokens | default(0) | int | format_number }}
- **估算总成本**: ${{ total_cost | round(2) }}
- **平均每次调用成本**: ${{ avg_cost_per_call | round(4) }}

## 按模型消耗统计
| 模型 | 调用次数 | 输入 Token | 输出 Token | 估算成本 |
|------|----------|------------|------------|----------|
{% for stat in model_stats %}
| {{ stat.model }} | {{ stat.call_count }} | {{ stat.input_tokens_est | int | format_number }} | {{ stat.output_tokens_est | int | int | format_number }} | ${{ stat.estimated_cost | round(2) }} |
{% endfor %}

## 诊断发现的问题
{% if diagnoses %}
{% for diag in diagnoses %}
### {{ diag.title }} ({{ diag.severity }})
- **描述**: {{ diag.description }}
- **建议**: {{ diag.suggestion }}
- **影响请求**: {{ diag.affected_request_ids | join(', ') }}
{% if diag.context %}
- **上下文**: {{ diag.context }}
{% endif %}
{% endfor %}
{% else %}
未发现显著问题。
{% endif %}

## 成本最高的 10 次请求
{% for req in top_expensive_requests %}
### 请求 ID: {{ req.request_id }}
- **时间**: {{ req.timestamp }}
- **模型**: {{ req.model }}
- **输入 Token**: {{ req.input_tokens_est }}
- **输出 Token**: {{ req.output_tokens_est }}
- **估算成本**: ${{ req.estimated_cost | round(4) }}
- **提示词结构预览**: {{ req.prompt_structure | truncate(200) }}
{% endfor %}

报告的价值 ：这份报告不仅是一份账单，更是一份 优化指南 。它直接告诉你：

1. 钱花在哪了 ：是按模型、按时间段的消耗分布。
2. 问题出在哪 ：具体的诊断条目，精确到请求 ID。
3. 怎么省 ：每一条诊断都附带了可操作的建议。

4. 部署、集成与实战避坑指南

4.1 如何集成到现有项目

集成方式取决于你现有的代码结构。以下是几种常见场景：

场景A：使用 requests 或 httpx 直接调用 API

import httpx
from claude_diagnostics import ClaudeDiagnosticsTransport, start_logging_consumer

# 1. 启动后台日志消费任务（只需一次）
log_queue = asyncio.Queue()
consumer_task = asyncio.create_task(start_logging_consumer(log_queue, db_path="claude_logs.db"))

# 2. 创建带有诊断传输层的客户端
async with httpx.AsyncClient(
    transport=ClaudeDiagnosticsTransport(log_queue=log_queue),
    timeout=30.0,
) as client:
    # 你的原有 API 调用代码
    response = await client.post(
        "https://api.anthropic.com/v1/messages",
        headers={"x-api-key": "your-key", "anthropic-version": "2023-06-01"},
        json={"model": "claude-3-sonnet-20240229", "max_tokens": 1024, "messages": [...]}
    )

场景B：使用 Anthropic 官方 Python SDK 官方 SDK 最终也是通过 HTTP 客户端发起请求。你可以通过猴子补丁（Monkey-patch）或在初始化时传入自定义的 HTTP 客户端来集成。

import anthropic
from claude_diagnostics import get_diagnostic_httpx_client

# 方法1：创建自定义客户端并传入（如果 SDK 支持）
client = anthropic.Anthropic(
    api_key="your-key",
    http_client=get_diagnostic_httpx_client() # 返回一个配置好的 httpx.AsyncClient
)

# 方法2：更通用的，装饰 SDK 的底层请求方法（需要查看 SDK 源码确定切入点）

场景C：在 Web 框架（如 FastAPI）中使用 在中间件（Middleware）中集成最为方便。你可以创建一个全局的诊断客户端，并在应用启动/关闭时管理日志消费者的生命周期。

4.2 常见问题与排查技巧

在开发和实际使用这个工具的过程中，我踩过不少坑，也总结了一些经验：

1. 性能开销 ：拦截和日志记录会引入额外开销。 解决方案 ：确保所有日志处理（尤其是 Token 计算）都是异步的，并且写入磁盘或数据库的操作使用缓冲队列，避免阻塞主线程。在我的测试中，对于单次 API 调用，额外延迟通常小于 50ms。

2. 日志数据量爆炸 ：如果 API 调用非常频繁，日志文件会快速增长。 解决方案 ：

   • 实现日志轮转（Rotating Logs），按大小或时间分割文件。
   • 在 SQLite 中，可以定期归档或清理旧数据（例如，只保留最近30天的详细日志，更早的数据只保留聚合统计）。
   • 考虑在生产环境中，将日志发送到更专业的可观测性平台（如 Datadog, Elasticsearch），本工具仅作为开发/调试期的深度诊断手段。

3. 误报与漏报 ：诊断规则不可能 100% 准确。 解决方案 ：

   • 为每条诊断规则设置可调节的敏感度参数（如相似度阈值、时间窗口），并允许用户通过配置文件调整。
   • 在报告中提供“标记为误报”或“忽略此类问题”的反馈机制，这些反馈可以用来优化规则。
   • 定期审查规则的有效性，对于经常产生误报的规则进行调优或暂时禁用。

4. 安全与隐私 ：日志中可能包含敏感的业务数据或用户信息。 解决方案 ：

   • 必须脱敏 ：在日志记录层，自动过滤或哈希化（Hash）可能包含敏感信息的字段，如 api_key 、 password 、 email 等。
   • 访问控制 ：生成的诊断报告应存储在安全的位置，并限制访问权限。
   • 合规性考虑 ：如果处理的是用户数据，需确保符合相关的数据保护法规（如 GDPR），考虑对日志进行匿名化处理。

5. 多进程/多线程环境 ：如果应用使用多进程，每个进程都有自己的日志队列和消费者，可能导致数据混乱。 解决方案 ：将日志队列和消费者进程设计为全局单例，或者使用进程安全的通信机制（如 Redis Pub/Sub）来集中日志，再由一个独立的消费者进程统一处理。

4.3 从诊断到优化：我的实战案例

工具建好后，我把它用在了自己的几个项目上，效果立竿见影：

• 案例一：自动化内容生成流水线 。诊断报告显示，一个用于生成产品描述的脚本，其系统提示词（包含品牌风格指南）长达 1200 Token，但每次调用都重复发送。 优化 ：我将风格指南提取为一个单独的文档，首次调用时通过文件上传传入，后续调用只需引用文档 ID，并将系统提示词精简为 150 Token 的核心指令。仅此一项，就让该流水线的成本降低了 65%。

• 案例二：聊天机器人中的“循环陷阱” 。规则引擎标记出一系列在 2 秒内发出的、相似度高达 95% 的请求。排查代码发现，前端在用户快速点击“重新生成”按钮时，没有做防抖处理，导致后端在极短时间内收到了多个相同请求。 优化 ：在前端增加了按钮防抖，在后端对相同会话和用户输入的请求增加了 5 秒的短期内存缓存。这个问题几乎消除了非必要的重复计算。

• 案例三：过高的 max_tokens 设置 。报告指出，许多请求的输出 Token 数远低于设置的 max_tokens （例如，设置了 4096，但平均输出只有 200）。这意味着我们为未使用的“容量”支付了潜在的心理和配置成本（虽然不影响直接计费，但可能影响模型生成策略）。 优化 ：根据历史输出长度的分布（P95），我动态调整了不同任务类型的 max_tokens 默认值，使其更贴合实际需求。

5. 总结与展望

构建这个 CLAUDE.md 诊断工具的过程，本质上是一次将“成本意识”工程化的实践。它让我从被动接收账单，转变为主动管理和理解每一次 API 调用的价值与损耗。工具本身并不复杂，但其带来的视角转变是革命性的。

我个人最深的体会是：优化往往始于可见性（Visibility） 。当你无法度量一件事时，你就无法管理它。这个工具提供的正是这种度量能力。它把模糊的“Token 消耗高”这个问题，拆解成了一个个具体的、可定位、可解决的子问题：是提示词冗长？是调用模式低效？还是参数配置不当？

对于想要尝试类似工具的朋友，我的建议是： 从最小可行产品（MVP）开始 。不必一开始就追求完美的规则引擎和漂亮的报告。可以先从最简单的日志拦截和 Token 计数做起，把数据收集起来。当你看着这些原始数据时，优化的方向自然会浮现。然后，再根据你遇到的具体问题，去编写一两条最有价值的诊断规则。工具会在解决实际问题的过程中，自然而然地成长和完善。

这个工具目前还是一个围绕我个人和团队需求构建的内部项目。未来，我考虑将其开源，并希望社区能一起贡献更多的诊断规则，覆盖更复杂的场景，比如流式响应（Streaming）的 Token 估算、多轮对话的上下文管理效率分析、以及结合业务指标（如用户满意度）的成本效益分析等。最终，我们希望的不只是“停止烧钱”，而是让每一分花在 AI API 上的钱，都产生最大的价值。