多 Agent Token 優化系列 pt.6：Agent 專責化實戰指南 — 打造精準高效的專家團隊

 1from enum import Enum
 2from dataclasses import dataclass, field
 3from typing import Optional
 4
 5class AgentRole(Enum):
 6    """Agent 角色類型"""
 7    # 分析類
 8    REQUIREMENTS_ANALYST = "requirements_analyst"
 9    DATA_ANALYST = "data_analyst"
10    SECURITY_ANALYST = "security_analyst"
11
12    # 設計類
13    ARCHITECT = "architect"
14    UI_DESIGNER = "ui_designer"
15    API_DESIGNER = "api_designer"
16
17    # 開發類
18    BACKEND_DEVELOPER = "backend_developer"
19    FRONTEND_DEVELOPER = "frontend_developer"
20    DATABASE_DEVELOPER = "database_developer"
21
22    # 品質類
23    CODE_REVIEWER = "code_reviewer"
24    SECURITY_REVIEWER = "security_reviewer"
25    PERFORMANCE_REVIEWER = "performance_reviewer"
26
27    # 文件類
28    DOC_WRITER = "doc_writer"
29    API_DOC_WRITER = "api_doc_writer"
30
31    # 測試類
32    TEST_DESIGNER = "test_designer"
33    TEST_IMPLEMENTER = "test_implementer"
34
35    # 協調類
36    ORCHESTRATOR = "orchestrator"
37    TASK_ROUTER = "task_router"
38
39@dataclass
40class AgentCapability:
41    """Agent 能力定義"""
42    role: AgentRole
43    description: str
44    core_skills: list[str]
45    tools_needed: list[str]
46    typical_output: str
47    recommended_model: str = "claude-sonnet-4-20250514"
48    max_system_tokens: int = 3000
49
50# 預定義的 Agent 能力庫
51AGENT_CAPABILITIES = {
52    AgentRole.REQUIREMENTS_ANALYST: AgentCapability(
53        role=AgentRole.REQUIREMENTS_ANALYST,
54        description="分析和整理使用者需求",
55        core_skills=["需求提取", "使用者故事撰寫", "驗收標準定義"],
56        tools_needed=["read_document"],
57        typical_output="結構化需求文件",
58        recommended_model="claude-sonnet-4-20250514",
59        max_system_tokens=2500
60    ),
61
62    AgentRole.BACKEND_DEVELOPER: AgentCapability(
63        role=AgentRole.BACKEND_DEVELOPER,
64        description="實作後端程式碼和 API",
65        core_skills=["Python/Go/Node.js", "API 設計", "資料庫操作"],
66        tools_needed=["read_file", "write_file", "run_command"],
67        typical_output="可執行的後端程式碼",
68        recommended_model="claude-sonnet-4-20250514",
69        max_system_tokens=3500
70    ),
71
72    AgentRole.CODE_REVIEWER: AgentCapability(
73        role=AgentRole.CODE_REVIEWER,
74        description="審查程式碼品質和最佳實踐",
75        core_skills=["程式碼分析", "問題識別", "改進建議"],
76        tools_needed=["read_file"],
77        typical_output="審查報告和建議",
78        recommended_model="claude-3-5-haiku-20241022",  # 輕量任務用 Haiku
79        max_system_tokens=2000
80    ),
81
82    AgentRole.ORCHESTRATOR: AgentCapability(
83        role=AgentRole.ORCHESTRATOR,
84        description="協調多個 Agent 完成複雜任務",
85        core_skills=["任務分解", "Agent 調度", "結果整合"],
86        tools_needed=["delegate_task"],
87        typical_output="整合後的最終結果",
88        recommended_model="claude-sonnet-4-20250514",
89        max_system_tokens=2000
90    ),
91}

  1import anthropic
  2from dataclasses import dataclass
  3from typing import Optional
  4
  5client = anthropic.Anthropic()
  6
  7@dataclass
  8class MinimalSystemPrompt:
  9    """最小化 System Prompt 結構"""
 10    role_definition: str       # 角色定義（必要）
 11    core_responsibility: str   # 核心職責（必要）
 12    output_format: str         # 輸出格式（必要）
 13    constraints: list[str]     # 限制條件（可選）
 14    examples: Optional[str] = None  # 範例（僅複雜任務需要）
 15
 16    def build(self) -> str:
 17        """建構最小化的 System Prompt"""
 18        parts = [
 19            f"# 角色\n{self.role_definition}",
 20            f"\n# 職責\n{self.core_responsibility}",
 21            f"\n# 輸出格式\n{self.output_format}",
 22        ]
 23
 24        if self.constraints:
 25            constraints_text = "\n".join(f"- {c}" for c in self.constraints)
 26            parts.append(f"\n# 限制\n{constraints_text}")
 27
 28        if self.examples:
 29            parts.append(f"\n# 範例\n{self.examples}")
 30
 31        return "\n".join(parts)
 32
 33    def token_estimate(self) -> int:
 34        """估算 token 數"""
 35        return len(self.build()) // 3
 36
 37
 38class SpecializedAgentFactory:
 39    """專責 Agent 工廠"""
 40
 41    # 預定義的精簡 System Prompts
 42    SPECIALIZED_PROMPTS = {
 43        AgentRole.REQUIREMENTS_ANALYST: MinimalSystemPrompt(
 44            role_definition="你是需求分析師，專精於理解和結構化使用者需求。",
 45            core_responsibility="""從使用者描述中提取：
 461. 功能需求（必須有的功能）
 472. 非功能需求（效能、安全、可用性）
 483. 限制條件
 494. 驗收標準""",
 50            output_format="""以 JSON 格式輸出：
 51{
 52    "functional_requirements": [...],
 53    "non_functional_requirements": [...],
 54    "constraints": [...],
 55    "acceptance_criteria": [...]
 56}""",
 57            constraints=[
 58                "只分析需求，不提供實作方案",
 59                "保持客觀，不加入假設",
 60                "不確定的需求標記為 [待確認]"
 61            ]
 62        ),
 63
 64        AgentRole.BACKEND_DEVELOPER: MinimalSystemPrompt(
 65            role_definition="你是後端開發者，專精於 Python/FastAPI 開發。",
 66            core_responsibility="根據需求和設計實作乾淨、可維護的後端程式碼。",
 67            output_format="""輸出完整可執行的程式碼：
 681. 必要的 import
 692. 完整的實作
 703. 基本的錯誤處理
 714. 簡短的使用說明""",
 72            constraints=[
 73                "遵循 PEP 8 風格",
 74                "函數單一職責",
 75                "加入類型提示"
 76            ]
 77        ),
 78
 79        AgentRole.CODE_REVIEWER: MinimalSystemPrompt(
 80            role_definition="你是程式碼審查員，專注於發現問題和提供改進建議。",
 81            core_responsibility="""審查程式碼的：
 821. 正確性
 832. 可讀性
 843. 效能
 854. 安全性""",
 86            output_format="""以列表形式輸出：
 87- 🔴 嚴重問題：[問題描述和建議]
 88- 🟡 建議改進：[改進點和原因]
 89- 🟢 做得好的地方：[肯定的部分]""",
 90            constraints=[
 91                "只審查，不修改程式碼",
 92                "每個問題都要有具體建議",
 93                "按嚴重程度排序"
 94            ]
 95        ),
 96
 97        AgentRole.DOC_WRITER: MinimalSystemPrompt(
 98            role_definition="你是技術文件撰寫者，專精於清晰易懂的文件。",
 99            core_responsibility="根據程式碼和設計撰寫技術文件。",
100            output_format="""Markdown 格式，包含：
1011. 概述
1022. 安裝/使用方法
1033. API 說明
1044. 範例""",
105            constraints=[
106                "使用簡潔清晰的語言",
107                "包含程式碼範例",
108                "面向目標讀者撰寫"
109            ]
110        ),
111
112        AgentRole.ORCHESTRATOR: MinimalSystemPrompt(
113            role_definition="你是任務協調者，負責分解任務並協調多個專家完成。",
114            core_responsibility="""分析任務後：
1151. 分解為子任務
1162. 指派給合適的專家
1173. 整合各專家的輸出""",
118            output_format="""JSON 格式的執行計畫：
119{
120    "subtasks": [
121        {"id": 1, "description": "...", "assigned_to": "agent_type"},
122        ...
123    ],
124    "execution_order": [1, 2, 3],
125    "dependencies": {"2": [1], "3": [1, 2]}
126}""",
127            constraints=[
128                "最小化子任務數量",
129                "明確定義依賴關係",
130                "每個子任務單一職責"
131            ]
132        ),
133    }
134
135    @classmethod
136    def create_agent(
137        cls,
138        role: AgentRole,
139        custom_constraints: Optional[list[str]] = None
140    ) -> "SpecializedAgent":
141        """創建專責 Agent"""
142        base_prompt = cls.SPECIALIZED_PROMPTS.get(role)
143        capability = AGENT_CAPABILITIES.get(role)
144
145        if not base_prompt or not capability:
146            raise ValueError(f"未定義的角色: {role}")
147
148        # 合併自訂限制
149        if custom_constraints:
150            prompt = MinimalSystemPrompt(
151                role_definition=base_prompt.role_definition,
152                core_responsibility=base_prompt.core_responsibility,
153                output_format=base_prompt.output_format,
154                constraints=base_prompt.constraints + custom_constraints,
155                examples=base_prompt.examples
156            )
157        else:
158            prompt = base_prompt
159
160        return SpecializedAgent(
161            role=role,
162            system_prompt=prompt.build(),
163            model=capability.recommended_model,
164            tools=capability.tools_needed
165        )
166
167
168class SpecializedAgent:
169    """專責 Agent"""
170
171    def __init__(
172        self,
173        role: AgentRole,
174        system_prompt: str,
175        model: str,
176        tools: list[str]
177    ):
178        self.role = role
179        self.system_prompt = system_prompt
180        self.model = model
181        self.tools = tools
182        self._stats = {
183            "calls": 0,
184            "total_input_tokens": 0,
185            "total_output_tokens": 0
186        }
187
188    def execute(self, task: str, context: str = "") -> str:
189        """執行任務"""
190        full_task = f"{context}\n\n任務：{task}" if context else task
191
192        response = client.messages.create(
193            model=self.model,
194            max_tokens=4096,
195            system=self.system_prompt,
196            messages=[{"role": "user", "content": full_task}]
197        )
198
199        # 更新統計
200        self._stats["calls"] += 1
201        self._stats["total_input_tokens"] += response.usage.input_tokens
202        self._stats["total_output_tokens"] += response.usage.output_tokens
203
204        return response.content[0].text
205
206    def get_stats(self) -> dict:
207        return {
208            **self._stats,
209            "system_prompt_tokens": len(self.system_prompt) // 3,
210            "model": self.model,
211            "tools_count": len(self.tools)
212        }
213
214
215# 使用範例
216if __name__ == "__main__":
217    # 創建專責 Agents
218    analyst = SpecializedAgentFactory.create_agent(AgentRole.REQUIREMENTS_ANALYST)
219    developer = SpecializedAgentFactory.create_agent(AgentRole.BACKEND_DEVELOPER)
220    reviewer = SpecializedAgentFactory.create_agent(AgentRole.CODE_REVIEWER)
221
222    print("專責 Agent System Prompt 大小：")
223    print(f"  需求分析師: ~{analyst.get_stats()['system_prompt_tokens']} tokens")
224    print(f"  後端開發者: ~{developer.get_stats()['system_prompt_tokens']} tokens")
225    print(f"  程式碼審查: ~{reviewer.get_stats()['system_prompt_tokens']} tokens")
226
227    # 對比通用 Agent
228    general_prompt_tokens = 16000
229    specialized_avg = (
230        analyst.get_stats()['system_prompt_tokens'] +
231        developer.get_stats()['system_prompt_tokens'] +
232        reviewer.get_stats()['system_prompt_tokens']
233    ) / 3
234
235    print(f"\n對比通用 Agent ({general_prompt_tokens} tokens):")
236    print(f"  專責 Agent 平均: ~{specialized_avg:.0f} tokens")
237    print(f"  節省: {(1 - specialized_avg/general_prompt_tokens) * 100:.1f}%")

  1from typing import Any
  2
  3# 完整的工具定義庫
  4TOOL_DEFINITIONS = {
  5    "read_file": {
  6        "name": "read_file",
  7        "description": "讀取檔案內容",
  8        "input_schema": {
  9            "type": "object",
 10            "properties": {
 11                "path": {"type": "string", "description": "檔案路徑"}
 12            },
 13            "required": ["path"]
 14        }
 15    },
 16    "write_file": {
 17        "name": "write_file",
 18        "description": "寫入檔案",
 19        "input_schema": {
 20            "type": "object",
 21            "properties": {
 22                "path": {"type": "string"},
 23                "content": {"type": "string"}
 24            },
 25            "required": ["path", "content"]
 26        }
 27    },
 28    "run_command": {
 29        "name": "run_command",
 30        "description": "執行系統命令",
 31        "input_schema": {
 32            "type": "object",
 33            "properties": {
 34                "command": {"type": "string"}
 35            },
 36            "required": ["command"]
 37        }
 38    },
 39    "query_database": {
 40        "name": "query_database",
 41        "description": "執行資料庫查詢",
 42        "input_schema": {
 43            "type": "object",
 44            "properties": {
 45                "query": {"type": "string"},
 46                "database": {"type": "string"}
 47            },
 48            "required": ["query"]
 49        }
 50    },
 51    "search_web": {
 52        "name": "search_web",
 53        "description": "搜尋網頁資訊",
 54        "input_schema": {
 55            "type": "object",
 56            "properties": {
 57                "query": {"type": "string"}
 58            },
 59            "required": ["query"]
 60        }
 61    },
 62    "delegate_task": {
 63        "name": "delegate_task",
 64        "description": "將任務委派給其他 Agent",
 65        "input_schema": {
 66            "type": "object",
 67            "properties": {
 68                "agent_type": {"type": "string"},
 69                "task": {"type": "string"},
 70                "context": {"type": "string"}
 71            },
 72            "required": ["agent_type", "task"]
 73        }
 74    },
 75    "analyze_code": {
 76        "name": "analyze_code",
 77        "description": "分析程式碼結構和依賴",
 78        "input_schema": {
 79            "type": "object",
 80            "properties": {
 81                "code": {"type": "string"},
 82                "language": {"type": "string"}
 83            },
 84            "required": ["code"]
 85        }
 86    },
 87    "run_tests": {
 88        "name": "run_tests",
 89        "description": "執行測試套件",
 90        "input_schema": {
 91            "type": "object",
 92            "properties": {
 93                "test_path": {"type": "string"},
 94                "coverage": {"type": "boolean"}
 95            },
 96            "required": ["test_path"]
 97        }
 98    }
 99}
100
101# Agent 角色到工具的映射（最小化配置）
102ROLE_TOOL_MAPPING = {
103    AgentRole.REQUIREMENTS_ANALYST: [
104        "read_file",  # 讀取需求文件
105    ],
106    AgentRole.DATA_ANALYST: [
107        "read_file",
108        "query_database",
109    ],
110    AgentRole.ARCHITECT: [
111        "read_file",
112        "search_web",  # 搜尋最佳實踐
113    ],
114    AgentRole.BACKEND_DEVELOPER: [
115        "read_file",
116        "write_file",
117        "run_command",
118    ],
119    AgentRole.FRONTEND_DEVELOPER: [
120        "read_file",
121        "write_file",
122        "run_command",
123    ],
124    AgentRole.CODE_REVIEWER: [
125        "read_file",  # 只需要讀取，不需要寫入
126        "analyze_code",
127    ],
128    AgentRole.SECURITY_REVIEWER: [
129        "read_file",
130        "analyze_code",
131    ],
132    AgentRole.DOC_WRITER: [
133        "read_file",
134        "write_file",
135    ],
136    AgentRole.TEST_DESIGNER: [
137        "read_file",
138    ],
139    AgentRole.TEST_IMPLEMENTER: [
140        "read_file",
141        "write_file",
142        "run_tests",
143    ],
144    AgentRole.ORCHESTRATOR: [
145        "delegate_task",
146    ],
147}
148
149
150class MinimalToolConfigurator:
151    """最小化工具配置器"""
152
153    def __init__(self):
154        self.tool_definitions = TOOL_DEFINITIONS
155        self.role_mapping = ROLE_TOOL_MAPPING
156
157    def get_tools_for_role(self, role: AgentRole) -> list[dict]:
158        """取得角色所需的最小工具集"""
159        tool_names = self.role_mapping.get(role, [])
160        return [
161            self.tool_definitions[name]
162            for name in tool_names
163            if name in self.tool_definitions
164        ]
165
166    def estimate_tools_tokens(self, role: AgentRole) -> int:
167        """估算工具定義的 token 數"""
168        import json
169        tools = self.get_tools_for_role(role)
170        tools_json = json.dumps(tools, ensure_ascii=False)
171        return len(tools_json) // 3
172
173    def compare_tool_overhead(self) -> dict:
174        """比較不同角色的工具開銷"""
175        # 計算全部工具的開銷
176        import json
177        all_tools_tokens = len(json.dumps(list(self.tool_definitions.values()), ensure_ascii=False)) // 3
178
179        results = {
180            "all_tools_tokens": all_tools_tokens,
181            "role_tools": {}
182        }
183
184        for role in AgentRole:
185            if role in self.role_mapping:
186                tokens = self.estimate_tools_tokens(role)
187                results["role_tools"][role.value] = {
188                    "tokens": tokens,
189                    "tools_count": len(self.role_mapping[role]),
190                    "savings_pct": f"{(1 - tokens/all_tools_tokens) * 100:.1f}%"
191                }
192
193        return results
194
195
196class ToolMinimizedAgent:
197    """工具最小化的 Agent"""
198
199    def __init__(
200        self,
201        role: AgentRole,
202        system_prompt: str,
203        model: str = "claude-sonnet-4-20250514"
204    ):
205        self.role = role
206        self.system_prompt = system_prompt
207        self.model = model
208
209        # 只配置必要的工具
210        configurator = MinimalToolConfigurator()
211        self.tools = configurator.get_tools_for_role(role)
212
213        self._tool_calls: list[dict] = []
214
215    def _handle_tool_call(self, tool_name: str, tool_input: dict) -> str:
216        """處理工具呼叫（示範實作）"""
217        # 實際應用中，這裡會呼叫真正的工具
218        self._tool_calls.append({"name": tool_name, "input": tool_input})
219
220        if tool_name == "read_file":
221            return f"[模擬] 讀取檔案 {tool_input.get('path')} 的內容..."
222        elif tool_name == "write_file":
223            return f"[模擬] 已寫入 {tool_input.get('path')}"
224        elif tool_name == "run_command":
225            return f"[模擬] 執行命令: {tool_input.get('command')}"
226        else:
227            return f"[模擬] 工具 {tool_name} 執行完成"
228
229    def execute(self, task: str, context: str = "") -> str:
230        """執行任務（支援工具呼叫）"""
231        messages = []
232
233        if context:
234            messages.append({"role": "user", "content": f"{context}\n\n任務：{task}"})
235        else:
236            messages.append({"role": "user", "content": task})
237
238        # 迭代處理工具呼叫
239        while True:
240            response = client.messages.create(
241                model=self.model,
242                max_tokens=4096,
243                system=self.system_prompt,
244                tools=self.tools if self.tools else None,
245                messages=messages
246            )
247
248            # 檢查是否有工具呼叫
249            tool_use_blocks = [
250                block for block in response.content
251                if block.type == "tool_use"
252            ]
253
254            if not tool_use_blocks:
255                # 沒有工具呼叫，返回文字回應
256                text_blocks = [
257                    block.text for block in response.content
258                    if hasattr(block, "text")
259                ]
260                return "\n".join(text_blocks)
261
262            # 處理工具呼叫
263            messages.append({"role": "assistant", "content": response.content})
264
265            tool_results = []
266            for tool_block in tool_use_blocks:
267                result = self._handle_tool_call(tool_block.name, tool_block.input)
268                tool_results.append({
269                    "type": "tool_result",
270                    "tool_use_id": tool_block.id,
271                    "content": result
272                })
273
274            messages.append({"role": "user", "content": tool_results})
275
276    def get_tool_stats(self) -> dict:
277        """取得工具使用統計"""
278        return {
279            "configured_tools": len(self.tools),
280            "tool_calls_made": len(self._tool_calls),
281            "tools_used": list(set(tc["name"] for tc in self._tool_calls))
282        }
283
284
285# 使用範例
286if __name__ == "__main__":
287    configurator = MinimalToolConfigurator()
288    comparison = configurator.compare_tool_overhead()
289
290    print("工具配置對比：")
291    print(f"全部工具: {comparison['all_tools_tokens']} tokens\n")
292
293    for role, stats in comparison["role_tools"].items():
294        print(f"{role}:")
295        print(f"  工具數: {stats['tools_count']}")
296        print(f"  Tokens: {stats['tokens']}")
297        print(f"  節省: {stats['savings_pct']}")

  1from enum import Enum
  2from dataclasses import dataclass
  3
  4class ModelTier(Enum):
  5    """模型層級"""
  6    FAST = "fast"       # Haiku - 快速便宜
  7    BALANCED = "balanced"  # Sonnet - 平衡
  8    POWERFUL = "powerful"  # Opus - 最強
  9
 10@dataclass
 11class ModelConfig:
 12    """模型配置"""
 13    tier: ModelTier
 14    model_id: str
 15    input_cost_per_1m: float
 16    output_cost_per_1m: float
 17    best_for: list[str]
 18
 19MODEL_CONFIGS = {
 20    ModelTier.FAST: ModelConfig(
 21        tier=ModelTier.FAST,
 22        model_id="claude-3-5-haiku-20241022",
 23        input_cost_per_1m=0.80,
 24        output_cost_per_1m=4.0,
 25        best_for=["分類", "摘要", "格式化", "簡單審查", "路由決策"]
 26    ),
 27    ModelTier.BALANCED: ModelConfig(
 28        tier=ModelTier.BALANCED,
 29        model_id="claude-sonnet-4-20250514",
 30        input_cost_per_1m=3.0,
 31        output_cost_per_1m=15.0,
 32        best_for=["程式碼生成", "分析", "一般任務", "協調"]
 33    ),
 34    ModelTier.POWERFUL: ModelConfig(
 35        tier=ModelTier.POWERFUL,
 36        model_id="claude-opus-4-20250514",
 37        input_cost_per_1m=15.0,
 38        output_cost_per_1m=75.0,
 39        best_for=["複雜架構", "創意設計", "深度分析", "關鍵決策"]
 40    ),
 41}
 42
 43# Agent 角色到模型層級的映射
 44ROLE_MODEL_MAPPING = {
 45    # 快速模型 - 簡單任務
 46    AgentRole.CODE_REVIEWER: ModelTier.FAST,
 47    AgentRole.DOC_WRITER: ModelTier.FAST,
 48    AgentRole.TEST_DESIGNER: ModelTier.FAST,
 49    AgentRole.TASK_ROUTER: ModelTier.FAST,
 50
 51    # 平衡模型 - 一般任務
 52    AgentRole.REQUIREMENTS_ANALYST: ModelTier.BALANCED,
 53    AgentRole.BACKEND_DEVELOPER: ModelTier.BALANCED,
 54    AgentRole.FRONTEND_DEVELOPER: ModelTier.BALANCED,
 55    AgentRole.DATA_ANALYST: ModelTier.BALANCED,
 56    AgentRole.ORCHESTRATOR: ModelTier.BALANCED,
 57    AgentRole.TEST_IMPLEMENTER: ModelTier.BALANCED,
 58
 59    # 強力模型 - 複雜任務
 60    AgentRole.ARCHITECT: ModelTier.POWERFUL,
 61    AgentRole.SECURITY_ANALYST: ModelTier.POWERFUL,
 62    AgentRole.SECURITY_REVIEWER: ModelTier.POWERFUL,
 63}
 64
 65
 66class ModelOptimizedAgentSystem:
 67    """模型優化的 Agent 系統"""
 68
 69    def __init__(self):
 70        self.agents: dict[AgentRole, SpecializedAgent] = {}
 71        self._cost_tracker = {
 72            "total_input_tokens": 0,
 73            "total_output_tokens": 0,
 74            "total_cost": 0.0,
 75            "by_tier": {tier: {"tokens": 0, "cost": 0.0} for tier in ModelTier}
 76        }
 77
 78    def get_model_for_role(self, role: AgentRole) -> str:
 79        """取得角色對應的模型"""
 80        tier = ROLE_MODEL_MAPPING.get(role, ModelTier.BALANCED)
 81        return MODEL_CONFIGS[tier].model_id
 82
 83    def get_model_tier(self, role: AgentRole) -> ModelTier:
 84        """取得角色對應的模型層級"""
 85        return ROLE_MODEL_MAPPING.get(role, ModelTier.BALANCED)
 86
 87    def register_agent(self, role: AgentRole, system_prompt: str):
 88        """註冊 Agent"""
 89        model = self.get_model_for_role(role)
 90
 91        # 取得最小化工具
 92        configurator = MinimalToolConfigurator()
 93        tools = configurator.get_tools_for_role(role)
 94
 95        self.agents[role] = SpecializedAgent(
 96            role=role,
 97            system_prompt=system_prompt,
 98            model=model,
 99            tools=[t["name"] for t in tools]
100        )
101
102    def execute(self, role: AgentRole, task: str, context: str = "") -> str:
103        """執行任務並追蹤成本"""
104        if role not in self.agents:
105            raise ValueError(f"Agent {role} 未註冊")
106
107        agent = self.agents[role]
108        tier = self.get_model_tier(role)
109        config = MODEL_CONFIGS[tier]
110
111        # 執行
112        result = agent.execute(task, context)
113
114        # 追蹤成本
115        stats = agent.get_stats()
116        input_tokens = stats["total_input_tokens"]
117        output_tokens = stats["total_output_tokens"]
118
119        cost = (
120            input_tokens * config.input_cost_per_1m / 1_000_000 +
121            output_tokens * config.output_cost_per_1m / 1_000_000
122        )
123
124        self._cost_tracker["total_input_tokens"] += input_tokens
125        self._cost_tracker["total_output_tokens"] += output_tokens
126        self._cost_tracker["total_cost"] += cost
127        self._cost_tracker["by_tier"][tier]["tokens"] += input_tokens + output_tokens
128        self._cost_tracker["by_tier"][tier]["cost"] += cost
129
130        return result
131
132    def get_cost_report(self) -> dict:
133        """取得成本報告"""
134        return {
135            "total_tokens": self._cost_tracker["total_input_tokens"] + self._cost_tracker["total_output_tokens"],
136            "total_cost": f"${self._cost_tracker['total_cost']:.4f}",
137            "by_tier": {
138                tier.value: {
139                    "tokens": stats["tokens"],
140                    "cost": f"${stats['cost']:.4f}"
141                }
142                for tier, stats in self._cost_tracker["by_tier"].items()
143            }
144        }
145
146    def estimate_savings_vs_single_model(self) -> dict:
147        """估算相比單一模型的節省"""
148        total_tokens = (
149            self._cost_tracker["total_input_tokens"] +
150            self._cost_tracker["total_output_tokens"]
151        )
152
153        if total_tokens == 0:
154            return {"error": "尚無執行記錄"}
155
156        # 假設全部使用各層級模型的成本
157        all_opus_cost = total_tokens * (15.0 + 75.0) / 2 / 1_000_000
158        all_sonnet_cost = total_tokens * (3.0 + 15.0) / 2 / 1_000_000
159        all_haiku_cost = total_tokens * (0.8 + 4.0) / 2 / 1_000_000
160
161        actual_cost = self._cost_tracker["total_cost"]
162
163        return {
164            "actual_cost": f"${actual_cost:.4f}",
165            "if_all_opus": f"${all_opus_cost:.4f}",
166            "if_all_sonnet": f"${all_sonnet_cost:.4f}",
167            "if_all_haiku": f"${all_haiku_cost:.4f}",
168            "savings_vs_opus": f"{(1 - actual_cost/all_opus_cost) * 100:.1f}%" if all_opus_cost > 0 else "N/A",
169            "savings_vs_sonnet": f"{(1 - actual_cost/all_sonnet_cost) * 100:.1f}%" if all_sonnet_cost > 0 else "N/A",
170        }

  1import anthropic
  2from dataclasses import dataclass, field
  3from typing import Optional, Any, Callable
  4from enum import Enum
  5import json
  6import time
  7
  8client = anthropic.Anthropic()
  9
 10@dataclass
 11class AgentExecutionResult:
 12    """Agent 執行結果"""
 13    role: AgentRole
 14    output: str
 15    input_tokens: int
 16    output_tokens: int
 17    model_used: str
 18    execution_time_ms: float
 19    cost: float
 20    success: bool
 21    error: Optional[str] = None
 22
 23@dataclass
 24class SpecializedAgentConfig:
 25    """專責 Agent 配置"""
 26    role: AgentRole
 27    system_prompt: str
 28    model_tier: ModelTier
 29    tools: list[str]
 30    max_tokens: int = 4096
 31
 32    # 可選的自訂處理器
 33    output_parser: Optional[Callable[[str], Any]] = None
 34    pre_processor: Optional[Callable[[str], str]] = None
 35
 36
 37class SpecializedAgentOrchestrator:
 38    """
 39    專責化 Agent 協調系統
 40
 41    整合：
 42    1. 最小化 System Prompt
 43    2. 最小化工具配置
 44    3. 模型差異化選擇
 45    4. 成本追蹤
 46    """
 47
 48    def __init__(self):
 49        self.agents: dict[AgentRole, SpecializedAgentConfig] = {}
 50        self.tool_configurator = MinimalToolConfigurator()
 51
 52        self._execution_history: list[AgentExecutionResult] = []
 53        self._stats = {
 54            "total_executions": 0,
 55            "total_tokens": 0,
 56            "total_cost": 0.0,
 57            "by_role": {},
 58            "by_model": {}
 59        }
 60
 61    def register_agent(self, config: SpecializedAgentConfig):
 62        """註冊專責 Agent"""
 63        self.agents[config.role] = config
 64        self._stats["by_role"][config.role.value] = {
 65            "executions": 0,
 66            "tokens": 0,
 67            "cost": 0.0
 68        }
 69
 70    def register_from_factory(self, role: AgentRole, custom_constraints: list[str] = None):
 71        """從工廠註冊 Agent"""
 72        # 取得預定義的 System Prompt
 73        base_prompt = SpecializedAgentFactory.SPECIALIZED_PROMPTS.get(role)
 74        if not base_prompt:
 75            raise ValueError(f"未定義角色 {role} 的 System Prompt")
 76
 77        # 建構 System Prompt
 78        if custom_constraints:
 79            prompt = MinimalSystemPrompt(
 80                role_definition=base_prompt.role_definition,
 81                core_responsibility=base_prompt.core_responsibility,
 82                output_format=base_prompt.output_format,
 83                constraints=base_prompt.constraints + custom_constraints,
 84                examples=base_prompt.examples
 85            )
 86        else:
 87            prompt = base_prompt
 88
 89        # 取得模型和工具配置
 90        model_tier = ROLE_MODEL_MAPPING.get(role, ModelTier.BALANCED)
 91        tools = [t["name"] for t in self.tool_configurator.get_tools_for_role(role)]
 92
 93        config = SpecializedAgentConfig(
 94            role=role,
 95            system_prompt=prompt.build(),
 96            model_tier=model_tier,
 97            tools=tools
 98        )
 99
100        self.register_agent(config)
101
102    def _get_model_id(self, tier: ModelTier) -> str:
103        """取得模型 ID"""
104        return MODEL_CONFIGS[tier].model_id
105
106    def _calculate_cost(self, tier: ModelTier, input_tokens: int, output_tokens: int) -> float:
107        """計算成本"""
108        config = MODEL_CONFIGS[tier]
109        return (
110            input_tokens * config.input_cost_per_1m / 1_000_000 +
111            output_tokens * config.output_cost_per_1m / 1_000_000
112        )
113
114    def execute_agent(
115        self,
116        role: AgentRole,
117        task: str,
118        context: str = ""
119    ) -> AgentExecutionResult:
120        """執行單個專責 Agent"""
121        if role not in self.agents:
122            return AgentExecutionResult(
123                role=role,
124                output="",
125                input_tokens=0,
126                output_tokens=0,
127                model_used="",
128                execution_time_ms=0,
129                cost=0,
130                success=False,
131                error=f"Agent {role.value} 未註冊"
132            )
133
134        config = self.agents[role]
135        model_id = self._get_model_id(config.model_tier)
136
137        # 預處理
138        if config.pre_processor:
139            task = config.pre_processor(task)
140
141        # 組合輸入
142        full_task = f"{context}\n\n任務：{task}" if context else task
143
144        # 取得工具定義
145        tools = self.tool_configurator.get_tools_for_role(role) if config.tools else None
146
147        start_time = time.time()
148
149        try:
150            kwargs = {
151                "model": model_id,
152                "max_tokens": config.max_tokens,
153                "system": config.system_prompt,
154                "messages": [{"role": "user", "content": full_task}]
155            }
156
157            if tools:
158                kwargs["tools"] = tools
159
160            response = client.messages.create(**kwargs)
161
162            # 提取文字輸出
163            text_output = "\n".join(
164                block.text for block in response.content
165                if hasattr(block, "text")
166            )
167
168            # 後處理
169            if config.output_parser:
170                text_output = config.output_parser(text_output)
171
172            execution_time = (time.time() - start_time) * 1000
173            cost = self._calculate_cost(
174                config.model_tier,
175                response.usage.input_tokens,
176                response.usage.output_tokens
177            )
178
179            result = AgentExecutionResult(
180                role=role,
181                output=text_output,
182                input_tokens=response.usage.input_tokens,
183                output_tokens=response.usage.output_tokens,
184                model_used=model_id,
185                execution_time_ms=execution_time,
186                cost=cost,
187                success=True
188            )
189
190        except Exception as e:
191            result = AgentExecutionResult(
192                role=role,
193                output="",
194                input_tokens=0,
195                output_tokens=0,
196                model_used=model_id,
197                execution_time_ms=(time.time() - start_time) * 1000,
198                cost=0,
199                success=False,
200                error=str(e)
201            )
202
203        # 更新統計
204        self._update_stats(result, config.model_tier)
205        self._execution_history.append(result)
206
207        return result
208
209    def _update_stats(self, result: AgentExecutionResult, tier: ModelTier):
210        """更新統計"""
211        self._stats["total_executions"] += 1
212        tokens = result.input_tokens + result.output_tokens
213        self._stats["total_tokens"] += tokens
214        self._stats["total_cost"] += result.cost
215
216        # 按角色
217        role_key = result.role.value
218        if role_key in self._stats["by_role"]:
219            self._stats["by_role"][role_key]["executions"] += 1
220            self._stats["by_role"][role_key]["tokens"] += tokens
221            self._stats["by_role"][role_key]["cost"] += result.cost
222
223        # 按模型
224        tier_key = tier.value
225        if tier_key not in self._stats["by_model"]:
226            self._stats["by_model"][tier_key] = {"executions": 0, "tokens": 0, "cost": 0.0}
227        self._stats["by_model"][tier_key]["executions"] += 1
228        self._stats["by_model"][tier_key]["tokens"] += tokens
229        self._stats["by_model"][tier_key]["cost"] += result.cost
230
231    def execute_pipeline(
232        self,
233        task: str,
234        pipeline: list[AgentRole],
235        context_builder: Optional[Callable[[list[AgentExecutionResult]], str]] = None
236    ) -> list[AgentExecutionResult]:
237        """
238        執行 Agent 管線
239
240        Args:
241            task: 原始任務
242            pipeline: Agent 執行順序
243            context_builder: 自訂 Context 建構器
244
245        Returns:
246            所有執行結果
247        """
248        results = []
249
250        for i, role in enumerate(pipeline):
251            print(f"\n{'='*50}")
252            print(f"Step {i+1}/{len(pipeline)}: {role.value}")
253            print(f"{'='*50}")
254
255            # 建構 Context
256            if context_builder and results:
257                context = context_builder(results)
258            elif results:
259                # 預設：使用上一個 Agent 的輸出摘要
260                last_result = results[-1]
261                context = f"[上一步 {last_result.role.value} 的輸出]\n{last_result.output[:2000]}..."
262            else:
263                context = ""
264
265            # 執行
266            result = self.execute_agent(role, task, context)
267            results.append(result)
268
269            # 輸出狀態
270            if result.success:
271                print(f"✅ 成功")
272                print(f"   模型: {result.model_used}")
273                print(f"   Tokens: {result.input_tokens + result.output_tokens}")
274                print(f"   成本: ${result.cost:.4f}")
275                print(f"   時間: {result.execution_time_ms:.0f}ms")
276            else:
277                print(f"❌ 失敗: {result.error}")
278
279        return results
280
281    def get_stats(self) -> dict:
282        """取得統計"""
283        return {
284            **self._stats,
285            "avg_tokens_per_execution": (
286                self._stats["total_tokens"] / self._stats["total_executions"]
287                if self._stats["total_executions"] > 0 else 0
288            ),
289            "avg_cost_per_execution": (
290                self._stats["total_cost"] / self._stats["total_executions"]
291                if self._stats["total_executions"] > 0 else 0
292            )
293        }
294
295    def get_optimization_report(self) -> str:
296        """取得優化報告"""
297        lines = [
298            "="*60,
299            "專責化 Agent 系統優化報告",
300            "="*60,
301            "",
302            f"總執行次數: {self._stats['total_executions']}",
303            f"總 Token 使用: {self._stats['total_tokens']:,}",
304            f"總成本: ${self._stats['total_cost']:.4f}",
305            "",
306            "按模型層級分布:",
307        ]
308
309        for tier, stats in self._stats["by_model"].items():
310            pct = stats["tokens"] / self._stats["total_tokens"] * 100 if self._stats["total_tokens"] > 0 else 0
311            lines.append(f"  {tier}: {stats['tokens']:,} tokens ({pct:.1f}%), ${stats['cost']:.4f}")
312
313        lines.extend([
314            "",
315            "按角色分布:",
316        ])
317
318        for role, stats in self._stats["by_role"].items():
319            if stats["executions"] > 0:
320                lines.append(f"  {role}: {stats['executions']} 次, {stats['tokens']:,} tokens, ${stats['cost']:.4f}")
321
322        # 估算節省
323        if self._stats["total_tokens"] > 0:
324            # 假設全部使用 Opus
325            all_opus_cost = self._stats["total_tokens"] * (15.0 + 75.0) / 2 / 1_000_000
326            savings = (all_opus_cost - self._stats["total_cost"]) / all_opus_cost * 100
327
328            lines.extend([
329                "",
330                "成本優化效果:",
331                f"  若全部使用 Opus: ${all_opus_cost:.4f}",
332                f"  實際成本: ${self._stats['total_cost']:.4f}",
333                f"  節省: {savings:.1f}%"
334            ])
335
336        lines.append("="*60)
337        return "\n".join(lines)
338
339
340# 使用範例
341def create_development_team() -> SpecializedAgentOrchestrator:
342    """創建開發團隊"""
343    orchestrator = SpecializedAgentOrchestrator()
344
345    # 註冊各專責 Agent
346    roles = [
347        AgentRole.REQUIREMENTS_ANALYST,
348        AgentRole.ARCHITECT,
349        AgentRole.BACKEND_DEVELOPER,
350        AgentRole.CODE_REVIEWER,
351        AgentRole.DOC_WRITER,
352    ]
353
354    for role in roles:
355        try:
356            orchestrator.register_from_factory(role)
357            print(f"✅ 註冊 {role.value}")
358        except ValueError as e:
359            print(f"⚠️ 跳過 {role.value}: {e}")
360
361    return orchestrator
362
363
364if __name__ == "__main__":
365    # 創建專責團隊
366    team = create_development_team()
367
368    # 定義開發任務
369    task = """
370    開發一個簡單的 URL 縮短服務：
371    - 接受長 URL，返回短碼
372    - 支援重定向
373    - 記錄訪問統計
374    """
375
376    # 執行開發管線
377    pipeline = [
378        AgentRole.REQUIREMENTS_ANALYST,
379        AgentRole.BACKEND_DEVELOPER,
380        AgentRole.CODE_REVIEWER,
381        AgentRole.DOC_WRITER,
382    ]
383
384    results = team.execute_pipeline(task, pipeline)
385
386    # 輸出報告
387    print("\n")
388    print(team.get_optimization_report())