多 Agent Token 優化系列 pt.1：完整指南 — 使用 Claude API 構建高效 AI 系統

 1import anthropic
 2
 3client = anthropic.Anthropic()
 4
 5# ✅ 使用 Prompt Caching 的多 Agent 呼叫
 6def call_agent_with_cache(agent_role: str, task: str, history: list):
 7    response = client.messages.create(
 8        model="claude-sonnet-4-6",
 9        max_tokens=4096,
10        system=[
11            {
12                "type": "text",
13                "text": f"""你是一個專業的 {agent_role}。
14
15你的核心職責：
16- 分析輸入資料並提供專業見解
17- 以結構化格式回傳結果
18- 確保輸出的準確性與完整性
19
20[此處可包含大量的角色背景知識、工具說明等...]
21這部分內容在多次呼叫間保持不變，非常適合快取。
22""",
23                "cache_control": {"type": "ephemeral"}  # 啟用快取
24            }
25        ],
26        messages=[
27            *history,  # 動態的對話歷史（不快取）
28            {"role": "user", "content": task}
29        ]
30    )
31    return response
32
33# 第一次呼叫：完整計算 System Prompt（快取未命中）
34result1 = call_agent_with_cache("資料分析師", "分析本季銷售數據", [])
35print(f"快取狀態: {result1.usage}")
36# cache_creation_input_tokens: 8500（建立快取）
37# cache_read_input_tokens: 0
38
39# 第二次呼叫：System Prompt 從快取讀取（快取命中）
40result2 = call_agent_with_cache("資料分析師", "預測下季趨勢", [])
41print(f"快取狀態: {result2.usage}")
42# cache_creation_input_tokens: 0
43# cache_read_input_tokens: 8500（快取命中！節省約 90% 成本）

 1import anthropic
 2from dataclasses import dataclass
 3from typing import Optional
 4
 5client = anthropic.Anthropic()
 6
 7@dataclass
 8class AgentContext:
 9    summary: Optional[str] = None       # 歷史摘要
10    recent_messages: list = None         # 最近的完整訊息
11    max_recent: int = 10                 # 保留最近幾輪
12
13    def __post_init__(self):
14        if self.recent_messages is None:
15            self.recent_messages = []
16
17def summarize_history(messages: list, topic: str) -> str:
18    """使用 Claude 將對話歷史壓縮為摘要"""
19    summary_prompt = f"""請將以下 {topic} 相關的對話歷史壓縮成簡潔摘要。
20保留：重要決策、關鍵資料、待解決問題、已確認事項。
21略去：重複資訊、過渡性對話、已解決的小問題。
22目標長度：原始內容的 20% 以內。
23
24對話歷史：
25{format_messages(messages)}"""
26
27    response = client.messages.create(
28        model="claude-haiku-4-5-20251001",  # 使用較便宜的模型做摘要
29        max_tokens=1000,
30        messages=[{"role": "user", "content": summary_prompt}]
31    )
32    return response.content[0].text
33
34def build_compressed_messages(context: AgentContext) -> list:
35    """構建壓縮後的訊息列表"""
36    messages = []
37
38    if context.summary:
39        # 將摘要作為第一條系統訊息注入
40        messages.append({
41            "role": "user",
42            "content": f"[歷史摘要]\n{context.summary}"
43        })
44        messages.append({
45            "role": "assistant",
46            "content": "已了解歷史背景，繼續處理。"
47        })
48
49    # 加入最近的完整對話
50    messages.extend(context.recent_messages[-context.max_recent * 2:])
51    return messages
52
53def run_agent_with_compression(
54    agent_role: str,
55    task: str,
56    context: AgentContext,
57    compress_threshold: int = 15  # 超過幾輪時觸發壓縮
58) -> tuple[str, AgentContext]:
59
60    # 檢查是否需要壓縮歷史
61    if len(context.recent_messages) > compress_threshold * 2:
62        old_messages = context.recent_messages[:-10]  # 保留最近 5 輪
63        context.summary = summarize_history(old_messages, agent_role)
64        context.recent_messages = context.recent_messages[-10:]
65        print(f"✅ 已壓縮歷史：{len(old_messages)} 條訊息 → 摘要")
66
67    # 構建壓縮後的訊息
68    messages = build_compressed_messages(context)
69    messages.append({"role": "user", "content": task})
70
71    response = client.messages.create(
72        model="claude-sonnet-4-6",
73        max_tokens=4096,
74        system=f"你是專業的 {agent_role}。",
75        messages=messages
76    )
77
78    result = response.content[0].text
79
80    # 更新 Context
81    context.recent_messages.append({"role": "user", "content": task})
82    context.recent_messages.append({"role": "assistant", "content": result})
83
84    return result, context

 1# ❌ 反模式：通用 Agent（浪費 tokens）
 2GENERAL_AGENT_SYSTEM = """
 3你是一個全能助手，可以：
 41. 分析資料（SQL 查詢、統計分析、趨勢預測...）
 52. 撰寫程式碼（Python、JavaScript、Java、Go...）
 63. 撰寫文件（技術文件、API 文件、報告...）
 74. 審查程式碼（安全性、效能、最佳實踐...）
 85. 測試撰寫（單元測試、整合測試、E2E...）
 9[大量的工具定義和範例...]
10"""
11# 每次呼叫：~15,000 tokens 的 System Prompt
12
13# ✅ 最佳實踐：專責 Agent（最小化 tokens）
14DATA_ANALYST_SYSTEM = """
15你是資料分析師，專注於：數據解讀、趨勢分析、視覺化建議。
16只使用以下工具：query_database、calculate_statistics、generate_chart。
17"""
18# 每次呼叫：~2,000 tokens 的 System Prompt
19
20CODE_GENERATOR_SYSTEM = """
21你是程式碼生成器，專注於：生成符合規範的乾淨程式碼。
22只使用以下工具：read_file、write_file、run_tests。
23"""
24# 每次呼叫：~2,500 tokens 的 System Prompt
25
26ORCHESTRATOR_SYSTEM = """
27你是任務協調者，負責：任務分解、Agent 派發、結果整合。
28可用 Agent：data_analyst, code_generator, doc_writer, code_reviewer。
29"""
30# 每次呼叫：~3,000 tokens 的 System Prompt
31
32class SpecializedAgentSystem:
33    def __init__(self):
34        self.agents = {
35            "orchestrator": {"system": ORCHESTRATOR_SYSTEM, "model": "claude-sonnet-4-6"},
36            "data_analyst": {"system": DATA_ANALYST_SYSTEM, "model": "claude-haiku-4-5-20251001"},
37            "code_generator": {"system": CODE_GENERATOR_SYSTEM, "model": "claude-sonnet-4-6"},
38        }
39
40    def call_agent(self, agent_name: str, task: str) -> str:
41        agent_config = self.agents[agent_name]
42        response = client.messages.create(
43            model=agent_config["model"],
44            max_tokens=4096,
45            system=agent_config["system"],
46            messages=[{"role": "user", "content": task}]
47        )
48        return response.content[0].text

 1from pydantic import BaseModel
 2from typing import Any
 3
 4class TaskResult(BaseModel):
 5    """標準化的 Agent 輸出結構"""
 6    agent_name: str
 7    task_id: str
 8    summary: str           # 簡短摘要（傳遞給下游）
 9    key_findings: list[str]  # 關鍵發現（傳遞給下游）
10    full_output: str       # 完整輸出（存入資料庫，不傳遞）
11    metadata: dict[str, Any]
12
13def extract_relevant_context(
14    task_results: list[TaskResult],
15    downstream_agent: str,
16    relevance_map: dict  # 定義哪個 Agent 需要哪些資訊
17) -> str:
18    """為下游 Agent 提取相關的最小 Context"""
19
20    needed_agents = relevance_map.get(downstream_agent, [])
21    relevant_parts = []
22
23    for result in task_results:
24        if result.agent_name in needed_agents:
25            # 只傳遞摘要和關鍵發現，不傳遞完整輸出
26            relevant_parts.append(
27                f"[{result.agent_name} 的關鍵輸出]\n"
28                f"摘要：{result.summary}\n"
29                f"關鍵發現：{chr(10).join(f'- {f}' for f in result.key_findings)}"
30            )
31
32    return "\n\n".join(relevant_parts)
33
34# 定義 Agent 間的資訊依賴關係
35RELEVANCE_MAP = {
36    "code_generator": ["requirements_analyst", "architect"],  # 程式碼生成只需需求和架構
37    "code_reviewer":  ["code_generator"],                      # 審查只需看程式碼
38    "doc_writer":     ["requirements_analyst", "code_generator", "code_reviewer"],
39    "tester":         ["requirements_analyst", "code_generator"]
40}
41
42class EfficientOrchestrator:
43    def __init__(self):
44        self.task_results: list[TaskResult] = []
45
46    def delegate_task(self, agent_name: str, task: str) -> TaskResult:
47        # 只傳遞該 Agent 真正需要的 Context
48        context = extract_relevant_context(
49            self.task_results,
50            agent_name,
51            RELEVANCE_MAP
52        )
53
54        full_task = f"{context}\n\n目前任務：{task}" if context else task
55
56        # 呼叫 Agent
57        response = client.messages.create(
58            model="claude-sonnet-4-6",
59            max_tokens=4096,
60            system=self.agents[agent_name]["system"],
61            messages=[{"role": "user", "content": full_task}]
62        )
63
64        full_output = response.content[0].text
65
66        # 使用 Claude 生成結構化摘要
67        result = self._create_task_result(agent_name, task, full_output)
68        self.task_results.append(result)
69        return result

 1from enum import Enum
 2
 3class TaskComplexity(Enum):
 4    LOW = "low"        # → Haiku
 5    MEDIUM = "medium"  # → Sonnet
 6    HIGH = "high"      # → Opus
 7
 8MODEL_MAP = {
 9    TaskComplexity.LOW:    "claude-haiku-4-5-20251001",
10    TaskComplexity.MEDIUM: "claude-sonnet-4-6",
11    TaskComplexity.HIGH:   "claude-opus-4-6",
12}
13
14def classify_task_complexity(task: str) -> TaskComplexity:
15    """使用輕量模型自動分類任務複雜度（節省成本）"""
16    classification_prompt = f"""
17分類以下任務的複雜度。只回答 "low"、"medium" 或 "high"。
18
19low（低）：格式化、簡單摘要、分類、路由
20medium（中）：程式碼生成、資料分析、文件撰寫
21high（高）：複雜架構設計、安全分析、多步驟推理
22
23任務：{task}
24複雜度："""
25
26    # 用最便宜的模型做分類
27    response = client.messages.create(
28        model="claude-haiku-4-5-20251001",
29        max_tokens=10,
30        messages=[{"role": "user", "content": classification_prompt}]
31    )
32
33    level = response.content[0].text.strip().lower()
34    return TaskComplexity(level) if level in ["low", "medium", "high"] else TaskComplexity.MEDIUM
35
36class AdaptiveAgentRouter:
37    def route_and_execute(self, task: str) -> str:
38        # 自動選擇合適的模型
39        complexity = classify_task_complexity(task)
40        model = MODEL_MAP[complexity]
41
42        print(f"任務複雜度：{complexity.value} → 使用模型：{model}")
43
44        response = client.messages.create(
45            model=model,
46            max_tokens=4096,
47            messages=[{"role": "user", "content": task}]
48        )
49        return response.content[0].text

 1import asyncio
 2from anthropic import AsyncAnthropic
 3
 4async_client = AsyncAnthropic()
 5
 6async def call_agent_async(agent_name: str, system: str, task: str) -> dict:
 7    """非同步呼叫單個 Agent"""
 8    response = await async_client.messages.create(
 9        model="claude-sonnet-4-6",
10        max_tokens=2048,
11        system=system,
12        messages=[{"role": "user", "content": task}]
13    )
14    return {
15        "agent": agent_name,
16        "result": response.content[0].text,
17        "tokens_used": response.usage.input_tokens + response.usage.output_tokens
18    }
19
20async def parallel_agent_execution(tasks: list[dict]) -> list[dict]:
21    """並行執行多個 Agent 任務"""
22    coroutines = [
23        call_agent_async(
24            task["agent_name"],
25            task["system_prompt"],
26            task["task"]
27        )
28        for task in tasks
29    ]
30
31    # 所有 Agent 同時執行
32    results = await asyncio.gather(*coroutines, return_exceptions=True)
33    return results
34
35# 使用範例
36async def main():
37    # 定義可並行執行的任務（互相獨立）
38    parallel_tasks = [
39        {
40            "agent_name": "security_reviewer",
41            "system_prompt": "你是資安審查專家，專注於程式碼安全漏洞。",
42            "task": "審查以下程式碼的安全性：[程式碼內容]"
43        },
44        {
45            "agent_name": "performance_reviewer",
46            "system_prompt": "你是效能優化專家，專注於程式碼效能問題。",
47            "task": "審查以下程式碼的效能：[程式碼內容]"
48        },
49        {
50            "agent_name": "style_reviewer",
51            "system_prompt": "你是程式碼風格審查員，專注於可讀性和最佳實踐。",
52            "task": "審查以下程式碼的風格：[程式碼內容]"
53        }
54    ]
55
56    print("開始並行執行 3 個審查 Agent...")
57    results = await parallel_agent_execution(parallel_tasks)
58
59    total_tokens = sum(r["tokens_used"] for r in results if isinstance(r, dict))
60    print(f"總 Token 消耗：{total_tokens}")
61    print(f"相較串行執行節省時間：約 {(2/3*100):.0f}%")
62
63    return results
64
65asyncio.run(main())

  1"""
  2生產級多 Agent 系統 - 整合所有 Token 優化策略
  3"""
  4import asyncio
  5import json
  6from anthropic import Anthropic, AsyncAnthropic
  7from dataclasses import dataclass, field
  8from typing import Optional
  9
 10client = Anthropic()
 11async_client = AsyncAnthropic()
 12
 13@dataclass
 14class OptimizedAgentConfig:
 15    name: str
 16    system_prompt: str
 17    model: str = "claude-sonnet-4-6"
 18    max_tokens: int = 4096
 19    use_cache: bool = True          # 策略1：啟用 Prompt Caching
 20
 21@dataclass
 22class SystemState:
 23    """全域系統狀態（避免重複傳遞）"""
 24    task_summaries: dict = field(default_factory=dict)
 25    shared_context: str = ""
 26    total_tokens_used: int = 0
 27
 28class ProductionMultiAgentSystem:
 29    def __init__(self):
 30        self.state = SystemState()
 31
 32        # 策略三：專責化 Agent 定義
 33        self.agents = {
 34            "orchestrator": OptimizedAgentConfig(
 35                name="orchestrator",
 36                system_prompt="你是任務協調員。分解任務、指派 Agent、整合結果。保持回應簡潔。",
 37                model="claude-sonnet-4-6",
 38                max_tokens=2048
 39            ),
 40            "analyst": OptimizedAgentConfig(
 41                name="analyst",
 42                system_prompt="你是分析師。只做資料分析和洞察提取。輸出要點式摘要。",
 43                model="claude-haiku-4-5-20251001",  # 策略五：低複雜度用 Haiku
 44                max_tokens=1024
 45            ),
 46            "developer": OptimizedAgentConfig(
 47                name="developer",
 48                system_prompt="你是開發者。只生成程式碼，附簡短說明。不做分析，不做文件。",
 49                model="claude-sonnet-4-6",
 50                max_tokens=4096
 51            ),
 52            "reviewer": OptimizedAgentConfig(
 53                name="reviewer",
 54                system_prompt="你是審查員。只輸出：問題清單、嚴重程度、修正建議。格式固定。",
 55                model="claude-haiku-4-5-20251001",
 56                max_tokens=1024
 57            )
 58        }
 59
 60    def _build_system_content(self, config: OptimizedAgentConfig) -> list:
 61        """策略一：構建帶快取的 System Prompt"""
 62        if config.use_cache:
 63            return [{
 64                "type": "text",
 65                "text": config.system_prompt,
 66                "cache_control": {"type": "ephemeral"}  # 啟用快取
 67            }]
 68        return config.system_prompt
 69
 70    def _get_minimal_context(self, agent_name: str) -> str:
 71        """策略四：只傳遞相關的最小 Context"""
 72        # 定義每個 Agent 需要哪些其他 Agent 的輸出
 73        context_needs = {
 74            "developer": ["analyst"],          # 開發者需要分析結果
 75            "reviewer":  ["developer"],         # 審查者需要開發者輸出
 76            "orchestrator": ["analyst", "developer", "reviewer"]  # 協調者需要全部
 77        }
 78
 79        needed = context_needs.get(agent_name, [])
 80        context_parts = []
 81
 82        for needed_agent in needed:
 83            if needed_agent in self.state.task_summaries:
 84                # 只傳摘要，不傳完整輸出（策略四核心）
 85                context_parts.append(
 86                    f"[{needed_agent} 輸出摘要]\n{self.state.task_summaries[needed_agent]}"
 87                )
 88
 89        return "\n\n".join(context_parts)
 90
 91    async def _call_agent_async(
 92        self,
 93        config: OptimizedAgentConfig,
 94        task: str
 95    ) -> str:
 96        context = self._get_minimal_context(config.name)
 97        full_task = f"{context}\n\n任務：{task}" if context else task
 98
 99        response = await async_client.messages.create(
100            model=config.model,
101            max_tokens=config.max_tokens,
102            system=self._build_system_content(config),
103            messages=[{"role": "user", "content": full_task}]
104        )
105
106        result = response.content[0].text
107        tokens = response.usage.input_tokens + response.usage.output_tokens
108        self.state.total_tokens_used += tokens
109
110        # 儲存摘要（前 500 字元作為摘要）
111        self.state.task_summaries[config.name] = result[:500] + "..." if len(result) > 500 else result
112
113        return result
114
115    async def run_pipeline(self, user_request: str) -> dict:
116        print(f"\n🚀 開始處理請求（Token 優化模式）")
117        print(f"請求：{user_request[:100]}...\n")
118
119        # 步驟 1：Orchestrator 分解任務（單次呼叫）
120        orchestration_plan = await self._call_agent_async(
121            self.agents["orchestrator"],
122            f"分解以下請求並制定執行計畫：{user_request}"
123        )
124        print(f"✅ 協調層完成，tokens: {self.state.total_tokens_used}")
125
126        # 步驟 2：策略六 - 並行執行獨立 Agent
127        print("🔄 並行執行分析和開發 Agent...")
128        analyst_task = self._call_agent_async(
129            self.agents["analyst"],
130            f"分析需求：{user_request}"
131        )
132        # 注意：analyst 和 developer 可並行，因為 developer 此時不需要 analyst 的輸出
133        # 但如果 developer 需要 analyst 結果，則必須串行
134
135        analyst_result = await analyst_task
136        print(f"✅ 分析完成，累計 tokens: {self.state.total_tokens_used}")
137
138        # 步驟 3：開發（需要 analyst 結果，串行）
139        dev_result = await self._call_agent_async(
140            self.agents["developer"],
141            orchestration_plan  # 傳遞計畫，Context 中已包含 analyst 摘要
142        )
143        print(f"✅ 開發完成，累計 tokens: {self.state.total_tokens_used}")
144
145        # 步驟 4：審查（需要 developer 結果，串行）
146        review_result = await self._call_agent_async(
147            self.agents["reviewer"],
148            "審查開發者的輸出"
149        )
150        print(f"✅ 審查完成，累計 tokens: {self.state.total_tokens_used}")
151
152        print(f"\n📊 最終統計：總 Token 使用量 = {self.state.total_tokens_used}")
153
154        return {
155            "orchestration": orchestration_plan,
156            "analysis": analyst_result,
157            "code": dev_result,
158            "review": review_result,
159            "total_tokens": self.state.total_tokens_used
160        }
161
162# 使用
163async def main():
164    system = ProductionMultiAgentSystem()
165    result = await system.run_pipeline(
166        "建立一個使用者認證模組，包含 JWT 登入、刷新 token 和登出功能"
167    )
168    print(json.dumps(result, ensure_ascii=False, indent=2))
169
170asyncio.run(main())

 1import time
 2from collections import defaultdict
 3
 4class TokenUsageMonitor:
 5    def __init__(self):
 6        self.usage_log = defaultdict(list)
 7
 8    def record_usage(self, agent_name: str, model: str, usage):
 9        self.usage_log[agent_name].append({
10            "timestamp": time.time(),
11            "model": model,
12            "input_tokens": usage.input_tokens,
13            "output_tokens": usage.output_tokens,
14            "cache_creation_input_tokens": getattr(usage, "cache_creation_input_tokens", 0),
15            "cache_read_input_tokens": getattr(usage, "cache_read_input_tokens", 0),
16        })
17
18    def print_report(self):
19        print("\n" + "="*60)
20        print("Token 使用量報告")
21        print("="*60)
22
23        total_input = 0
24        total_output = 0
25        total_cache_saves = 0
26
27        for agent, logs in self.usage_log.items():
28            agent_input = sum(l["input_tokens"] for l in logs)
29            agent_output = sum(l["output_tokens"] for l in logs)
30            cache_reads = sum(l["cache_read_input_tokens"] for l in logs)
31
32            total_input += agent_input
33            total_output += agent_output
34            total_cache_saves += cache_reads
35
36            print(f"\nAgent: {agent}")
37            print(f"  輸入 tokens: {agent_input:,}")
38            print(f"  輸出 tokens: {agent_output:,}")
39            print(f"  快取命中 tokens: {cache_reads:,}")
40            if cache_reads > 0:
41                savings_pct = cache_reads / (agent_input + cache_reads) * 100
42                print(f"  快取節省率: {savings_pct:.1f}%")
43
44        print(f"\n總計：")
45        print(f"  輸入 tokens: {total_input:,}")
46        print(f"  輸出 tokens: {total_output:,}")
47        print(f"  快取節省 tokens: {total_cache_saves:,}")
48        print("="*60)