context-profiler/src/context_profiler/analyzers/token_counter.py at c8b8fff20e0ca270c18805fdd97c14ce04674ba5 · yanpgwang/context-profiler · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
"""Token counting analyzer — breakdown by role, content type, and tool name."""

from __future__ import annotations

from collections import defaultdict

from context_profiler.analyzers.base import AnalyzerResult, BaseAnalyzer
from context_profiler.models import APIRequest, BlockType, Role
from context_profiler.pricing import estimate_cost


class TokenCounterAnalyzer(BaseAnalyzer):

    @property
    def name(self) -> str:
        return "token_counter"

    def analyze(self, request: APIRequest) -> AnalyzerResult:
        by_role: dict[str, int] = defaultdict(int)
        by_content_type: dict[str, int] = defaultdict(int)
        by_tool_name: dict[str, int] = defaultdict(int)
        by_tool_name_input: dict[str, int] = defaultdict(int)
        by_tool_name_result: dict[str, int] = defaultdict(int)
        per_message: list[dict] = []
        total_tokens = 0

        for msg in request.messages:
            msg_tokens = msg.total_tokens
            total_tokens += msg_tokens
            by_role[msg.role.value] += msg_tokens

            for block in msg.blocks:
                by_content_type[block.block_type.value] += block.token_count
                if block.tool_name:
                    by_tool_name[block.tool_name] += block.token_count
                    if block.block_type == BlockType.TOOL_USE:
                        by_tool_name_input[block.tool_name] += block.token_count
                    elif block.block_type == BlockType.TOOL_RESULT:
                        by_tool_name_result[block.tool_name] += block.token_count

            per_message.append({
                "index": msg.index,
                "role": msg.role.value,
                "tokens": msg_tokens,
                "block_types": [b.block_type.value for b in msg.blocks],
            })

        tool_def_tokens = request.tool_definition_tokens
        total_tokens += tool_def_tokens

        top_messages = sorted(per_message, key=lambda x: x["tokens"], reverse=True)[:10]
        top_tools = sorted(by_tool_name.items(), key=lambda x: x[1], reverse=True)[:10]
        top_tools_input = sorted(by_tool_name_input.items(), key=lambda x: x[1], reverse=True)[:10]
        top_tools_result = sorted(by_tool_name_result.items(), key=lambda x: x[1], reverse=True)[:10]

        tool_defs_detail = [
            {"name": t.name, "tokens": t.token_count}
            for t in sorted(request.tools, key=lambda t: t.token_count, reverse=True)
        ]

        tool_use_tokens = by_content_type.get("tool_use", 0)
        tool_result_tokens = by_content_type.get("tool_result", 0)

        cost = estimate_cost(
            input_tokens=total_tokens,
            output_tokens=0,
            model=request.model,
        )

        summary = {
            "total_input_tokens": total_tokens,
            "message_tokens": total_tokens - tool_def_tokens,
            "tool_definition_tokens": tool_def_tokens,
            "system_prompt_tokens": request.system_prompt_tokens,
            "source_format": request.source_format,
            "model": request.model,
            "by_role": dict(by_role),
            "by_content_type": dict(by_content_type),
            "tool_use_tokens": tool_use_tokens,
            "tool_result_tokens": tool_result_tokens,
            "top_tools_by_tokens": top_tools,
            "top_tools_by_input_tokens": top_tools_input,
            "top_tools_by_result_tokens": top_tools_result,
            "tool_definitions": tool_defs_detail,
        }

        if cost is not None:
            summary["cost"] = cost

        warnings = []
        if tool_def_tokens > total_tokens * 0.3:
            warnings.append(
                f"Tool definitions consume {tool_def_tokens:,} tokens "
                f"({tool_def_tokens / total_tokens * 100:.1f}% of total)"
            )
        if request.system_prompt_tokens > total_tokens * 0.3:
            warnings.append(
                f"System prompt consumes {request.system_prompt_tokens:,} tokens "
                f"({request.system_prompt_tokens / total_tokens * 100:.1f}% of total)"
            )

        return AnalyzerResult(
            analyzer_name=self.name,
            summary=summary,
            details=top_messages,
            warnings=warnings,
        )