多 Agent Token 優化系列 pt.5：選擇性 Context 傳遞 — 打造高效協作系統

  1from dataclasses import dataclass, field
  2from typing import Optional
  3from enum import Enum
  4
  5class ContextType(Enum):
  6    """Context 類型"""
  7    FULL_OUTPUT = "full_output"       # 完整輸出
  8    SUMMARY = "summary"               # 摘要
  9    KEY_FINDINGS = "key_findings"     # 關鍵發現
 10    STRUCTURED_DATA = "structured"    # 結構化資料
 11    METADATA = "metadata"             # 元資料
 12    NONE = "none"                     # 不需要
 13
 14@dataclass
 15class ContextDependency:
 16    """Context 依賴定義"""
 17    source_agent: str                 # 來源 Agent
 18    context_type: ContextType         # 需要的 Context 類型
 19    required_fields: list[str] = field(default_factory=list)  # 需要的特定欄位
 20    max_tokens: Optional[int] = None  # 最大 token 數限制
 21    priority: int = 1                 # 優先級（用於裁剪）
 22
 23@dataclass
 24class AgentContextSpec:
 25    """Agent 的 Context 規格"""
 26    agent_name: str
 27    dependencies: list[ContextDependency] = field(default_factory=list)
 28    requires_original_task: bool = True
 29    requires_shared_state: bool = False
 30    max_total_context: int = 8000     # 最大總 Context
 31
 32# 定義依賴關係圖
 33CONTEXT_DEPENDENCY_GRAPH = {
 34    "orchestrator": AgentContextSpec(
 35        agent_name="orchestrator",
 36        dependencies=[],  # Orchestrator 接收原始請求
 37        requires_original_task=True,
 38        max_total_context=4000
 39    ),
 40
 41    "requirements_analyst": AgentContextSpec(
 42        agent_name="requirements_analyst",
 43        dependencies=[],  # 只需要原始任務
 44        requires_original_task=True,
 45        max_total_context=6000
 46    ),
 47
 48    "architect": AgentContextSpec(
 49        agent_name="architect",
 50        dependencies=[
 51            ContextDependency(
 52                source_agent="requirements_analyst",
 53                context_type=ContextType.KEY_FINDINGS,
 54                required_fields=["functional_requirements", "non_functional_requirements"],
 55                max_tokens=1500
 56            )
 57        ],
 58        requires_original_task=True,
 59        max_total_context=8000
 60    ),
 61
 62    "developer": AgentContextSpec(
 63        agent_name="developer",
 64        dependencies=[
 65            ContextDependency(
 66                source_agent="requirements_analyst",
 67                context_type=ContextType.SUMMARY,
 68                max_tokens=500
 69            ),
 70            ContextDependency(
 71                source_agent="architect",
 72                context_type=ContextType.STRUCTURED_DATA,
 73                required_fields=["architecture_decision", "tech_stack", "api_design"],
 74                max_tokens=2000
 75            )
 76        ],
 77        requires_original_task=False,  # 不需要原始任務
 78        max_total_context=10000
 79    ),
 80
 81    "code_reviewer": AgentContextSpec(
 82        agent_name="code_reviewer",
 83        dependencies=[
 84            ContextDependency(
 85                source_agent="developer",
 86                context_type=ContextType.FULL_OUTPUT,  # 需要完整程式碼
 87                max_tokens=5000
 88            ),
 89            ContextDependency(
 90                source_agent="architect",
 91                context_type=ContextType.SUMMARY,
 92                required_fields=["coding_standards"],
 93                max_tokens=500
 94            )
 95        ],
 96        requires_original_task=False,
 97        max_total_context=8000
 98    ),
 99
100    "doc_writer": AgentContextSpec(
101        agent_name="doc_writer",
102        dependencies=[
103            ContextDependency(
104                source_agent="requirements_analyst",
105                context_type=ContextType.KEY_FINDINGS,
106                max_tokens=800
107            ),
108            ContextDependency(
109                source_agent="architect",
110                context_type=ContextType.SUMMARY,
111                max_tokens=600
112            ),
113            ContextDependency(
114                source_agent="developer",
115                context_type=ContextType.METADATA,
116                required_fields=["file_list", "api_endpoints", "function_signatures"],
117                max_tokens=1000
118            ),
119            ContextDependency(
120                source_agent="code_reviewer",
121                context_type=ContextType.SUMMARY,
122                max_tokens=400
123            )
124        ],
125        requires_original_task=True,
126        max_total_context=6000
127    )
128}

  1from pydantic import BaseModel, Field
  2from typing import Optional, Any
  3from enum import Enum
  4
  5class OutputSection(Enum):
  6    """輸出區段類型"""
  7    SUMMARY = "summary"
  8    KEY_FINDINGS = "key_findings"
  9    DETAILED_ANALYSIS = "detailed_analysis"
 10    RECOMMENDATIONS = "recommendations"
 11    STRUCTURED_DATA = "structured_data"
 12    METADATA = "metadata"
 13    RAW_OUTPUT = "raw_output"
 14
 15class AgentOutput(BaseModel):
 16    """標準化 Agent 輸出"""
 17    agent_name: str
 18    task_id: str
 19
 20    # 必要的摘要區段（永遠保留）
 21    summary: str = Field(
 22        description="簡短摘要，100-200 字",
 23        max_length=500
 24    )
 25
 26    # 關鍵發現（高優先級保留）
 27    key_findings: list[str] = Field(
 28        default_factory=list,
 29        description="關鍵發現清單，每項 50 字以內"
 30    )
 31
 32    # 結構化資料（可選擇性提取欄位）
 33    structured_data: dict[str, Any] = Field(
 34        default_factory=dict,
 35        description="結構化資料，便於下游精確提取"
 36    )
 37
 38    # 詳細內容（通常不傳遞給下游）
 39    detailed_output: Optional[str] = Field(
 40        default=None,
 41        description="完整詳細輸出，僅在需要時傳遞"
 42    )
 43
 44    # 元資料（輕量級資訊）
 45    metadata: dict[str, Any] = Field(
 46        default_factory=dict,
 47        description="執行元資料：時間、tokens、狀態等"
 48    )
 49
 50    def get_section(self, section: OutputSection, max_tokens: Optional[int] = None) -> str:
 51        """取得特定區段，可選擇性截斷"""
 52        content = ""
 53
 54        if section == OutputSection.SUMMARY:
 55            content = self.summary
 56        elif section == OutputSection.KEY_FINDINGS:
 57            content = "\n".join(f"- {f}" for f in self.key_findings)
 58        elif section == OutputSection.STRUCTURED_DATA:
 59            import json
 60            content = json.dumps(self.structured_data, ensure_ascii=False, indent=2)
 61        elif section == OutputSection.DETAILED_ANALYSIS:
 62            content = self.detailed_output or ""
 63        elif section == OutputSection.METADATA:
 64            import json
 65            content = json.dumps(self.metadata, ensure_ascii=False)
 66        elif section == OutputSection.RAW_OUTPUT:
 67            content = self.detailed_output or self.summary
 68
 69        # 截斷處理
 70        if max_tokens and len(content) > max_tokens * 4:  # 粗略估算
 71            content = content[:max_tokens * 4] + "\n...[已截斷]"
 72
 73        return content
 74
 75    def to_minimal_context(self) -> str:
 76        """轉換為最小 Context（僅摘要和關鍵發現）"""
 77        parts = [f"[{self.agent_name} 輸出摘要]", self.summary]
 78        if self.key_findings:
 79            parts.append("\n關鍵發現：")
 80            parts.extend(f"- {f}" for f in self.key_findings[:5])
 81        return "\n".join(parts)
 82
 83    def to_full_context(self) -> str:
 84        """轉換為完整 Context"""
 85        parts = [
 86            f"[{self.agent_name} 完整輸出]",
 87            f"\n摘要：{self.summary}",
 88        ]
 89        if self.key_findings:
 90            parts.append("\n關鍵發現：")
 91            parts.extend(f"- {f}" for f in self.key_findings)
 92        if self.detailed_output:
 93            parts.append(f"\n詳細內容：\n{self.detailed_output}")
 94        return "\n".join(parts)
 95
 96
 97# 特定 Agent 的輸出格式
 98class RequirementsAnalystOutput(AgentOutput):
 99    """需求分析師輸出"""
100    structured_data: dict = Field(default_factory=lambda: {
101        "functional_requirements": [],
102        "non_functional_requirements": [],
103        "constraints": [],
104        "assumptions": [],
105        "user_stories": []
106    })
107
108class ArchitectOutput(AgentOutput):
109    """架構師輸出"""
110    structured_data: dict = Field(default_factory=lambda: {
111        "architecture_decision": "",
112        "tech_stack": {},
113        "api_design": [],
114        "data_model": {},
115        "coding_standards": [],
116        "deployment_strategy": ""
117    })
118
119class DeveloperOutput(AgentOutput):
120    """開發者輸出"""
121    structured_data: dict = Field(default_factory=lambda: {
122        "file_list": [],
123        "api_endpoints": [],
124        "function_signatures": [],
125        "dependencies": [],
126        "code_files": {}
127    })
128
129class CodeReviewerOutput(AgentOutput):
130    """程式碼審查者輸出"""
131    structured_data: dict = Field(default_factory=lambda: {
132        "issues": [],
133        "suggestions": [],
134        "security_concerns": [],
135        "performance_notes": [],
136        "approval_status": ""
137    })

  1import anthropic
  2import json
  3from typing import Type
  4
  5client = anthropic.Anthropic()
  6
  7class StructuredOutputAgent:
  8    """
  9    強制結構化輸出的 Agent
 10
 11    確保輸出符合標準格式，便於下游選擇性提取
 12    """
 13
 14    OUTPUT_FORMAT_PROMPT = """
 15你必須以以下 JSON 格式輸出結果：
 16
 17{{
 18    "summary": "簡短摘要（100-200字）",
 19    "key_findings": ["發現1", "發現2", ...],
 20    "structured_data": {{
 21        // 根據任務類型的結構化資料
 22        {structured_fields}
 23    }},
 24    "detailed_output": "完整詳細內容（可選）"
 25}}
 26
 27重要：只輸出 JSON，不要其他說明。
 28"""
 29
 30    def __init__(
 31        self,
 32        agent_name: str,
 33        system_prompt: str,
 34        output_class: Type[AgentOutput] = AgentOutput,
 35        model: str = "claude-sonnet-4-20250514"
 36    ):
 37        self.agent_name = agent_name
 38        self.system_prompt = system_prompt
 39        self.output_class = output_class
 40        self.model = model
 41
 42    def _get_structured_fields_hint(self) -> str:
 43        """取得結構化欄位提示"""
 44        if self.output_class == RequirementsAnalystOutput:
 45            return """
 46        "functional_requirements": ["需求1", "需求2"],
 47        "non_functional_requirements": ["效能要求", "安全要求"],
 48        "constraints": ["限制1"],
 49        "user_stories": ["作為...我想要...以便..."]
 50"""
 51        elif self.output_class == ArchitectOutput:
 52            return """
 53        "architecture_decision": "架構決策說明",
 54        "tech_stack": {"backend": "...", "frontend": "...", "database": "..."},
 55        "api_design": [{"endpoint": "/api/...", "method": "GET", "description": "..."}],
 56        "coding_standards": ["標準1", "標準2"]
 57"""
 58        elif self.output_class == DeveloperOutput:
 59            return """
 60        "file_list": ["file1.py", "file2.py"],
 61        "api_endpoints": ["/api/endpoint1", "/api/endpoint2"],
 62        "function_signatures": ["def func1(a: int) -> str", ...],
 63        "code_files": {"filename.py": "code content..."}
 64"""
 65        elif self.output_class == CodeReviewerOutput:
 66            return """
 67        "issues": [{"severity": "high/medium/low", "description": "..."}],
 68        "suggestions": ["建議1", "建議2"],
 69        "security_concerns": ["安全問題1"],
 70        "approval_status": "approved/needs_revision/rejected"
 71"""
 72        return "{}"
 73
 74    def execute(self, task: str, context: str = "") -> AgentOutput:
 75        """執行任務並返回結構化輸出"""
 76        format_prompt = self.OUTPUT_FORMAT_PROMPT.format(
 77            structured_fields=self._get_structured_fields_hint()
 78        )
 79
 80        full_prompt = f"{context}\n\n任務：{task}\n\n{format_prompt}" if context else f"任務：{task}\n\n{format_prompt}"
 81
 82        response = client.messages.create(
 83            model=self.model,
 84            max_tokens=4096,
 85            system=self.system_prompt,
 86            messages=[{"role": "user", "content": full_prompt}]
 87        )
 88
 89        # 解析輸出
 90        output_text = response.content[0].text
 91        try:
 92            # 處理可能的 markdown 程式碼區塊
 93            if "```json" in output_text:
 94                output_text = output_text.split("```json")[1].split("```")[0]
 95            elif "```" in output_text:
 96                output_text = output_text.split("```")[1].split("```")[0]
 97
 98            data = json.loads(output_text.strip())
 99
100            return self.output_class(
101                agent_name=self.agent_name,
102                task_id="",  # 由呼叫者設定
103                summary=data.get("summary", ""),
104                key_findings=data.get("key_findings", []),
105                structured_data=data.get("structured_data", {}),
106                detailed_output=data.get("detailed_output"),
107                metadata={
108                    "input_tokens": response.usage.input_tokens,
109                    "output_tokens": response.usage.output_tokens
110                }
111            )
112        except (json.JSONDecodeError, KeyError) as e:
113            # 降級處理：將原始輸出作為 detailed_output
114            return self.output_class(
115                agent_name=self.agent_name,
116                task_id="",
117                summary=output_text[:200] + "..." if len(output_text) > 200 else output_text,
118                key_findings=[],
119                structured_data={},
120                detailed_output=output_text,
121                metadata={"parse_error": str(e)}
122            )

  1import anthropic
  2from dataclasses import dataclass
  3from typing import Optional
  4
  5client = anthropic.Anthropic()
  6
  7@dataclass
  8class ExtractionRequest:
  9    """提取請求"""
 10    source_output: AgentOutput
 11    target_agent: str
 12    dependency: ContextDependency
 13
 14class ContextExtractor:
 15    """
 16    Context 提取器
 17
 18    根據依賴規格從 Agent 輸出中提取所需資訊
 19    """
 20
 21    def __init__(self, use_llm_extraction: bool = True):
 22        """
 23        Args:
 24            use_llm_extraction: 是否使用 LLM 進行智能提取（更準確但有成本）
 25        """
 26        self.use_llm_extraction = use_llm_extraction
 27
 28    def extract(self, request: ExtractionRequest) -> str:
 29        """
 30        從來源輸出提取目標 Agent 所需的 Context
 31        """
 32        source = request.source_output
 33        dep = request.dependency
 34
 35        # 根據 Context 類型選擇提取策略
 36        if dep.context_type == ContextType.NONE:
 37            return ""
 38
 39        elif dep.context_type == ContextType.SUMMARY:
 40            return self._extract_summary(source, dep)
 41
 42        elif dep.context_type == ContextType.KEY_FINDINGS:
 43            return self._extract_key_findings(source, dep)
 44
 45        elif dep.context_type == ContextType.STRUCTURED_DATA:
 46            return self._extract_structured_data(source, dep)
 47
 48        elif dep.context_type == ContextType.METADATA:
 49            return self._extract_metadata(source, dep)
 50
 51        elif dep.context_type == ContextType.FULL_OUTPUT:
 52            return self._extract_full_output(source, dep)
 53
 54        return source.to_minimal_context()
 55
 56    def _extract_summary(self, source: AgentOutput, dep: ContextDependency) -> str:
 57        """提取摘要"""
 58        content = f"[{source.agent_name} 摘要]\n{source.summary}"
 59        return self._truncate(content, dep.max_tokens)
 60
 61    def _extract_key_findings(self, source: AgentOutput, dep: ContextDependency) -> str:
 62        """提取關鍵發現"""
 63        parts = [f"[{source.agent_name} 關鍵發現]"]
 64
 65        # 如果指定了特定欄位，只提取那些欄位
 66        if dep.required_fields:
 67            for field in dep.required_fields:
 68                if field in source.structured_data:
 69                    value = source.structured_data[field]
 70                    if isinstance(value, list):
 71                        parts.append(f"\n{field}:")
 72                        parts.extend(f"  - {item}" for item in value[:10])
 73                    else:
 74                        parts.append(f"\n{field}: {value}")
 75        else:
 76            # 提取所有關鍵發現
 77            for finding in source.key_findings[:10]:
 78                parts.append(f"- {finding}")
 79
 80        content = "\n".join(parts)
 81        return self._truncate(content, dep.max_tokens)
 82
 83    def _extract_structured_data(self, source: AgentOutput, dep: ContextDependency) -> str:
 84        """提取結構化資料"""
 85        import json
 86
 87        if dep.required_fields:
 88            # 只提取指定欄位
 89            extracted = {}
 90            for field in dep.required_fields:
 91                if field in source.structured_data:
 92                    extracted[field] = source.structured_data[field]
 93            data = extracted
 94        else:
 95            data = source.structured_data
 96
 97        content = f"[{source.agent_name} 結構化資料]\n{json.dumps(data, ensure_ascii=False, indent=2)}"
 98        return self._truncate(content, dep.max_tokens)
 99
100    def _extract_metadata(self, source: AgentOutput, dep: ContextDependency) -> str:
101        """提取元資料"""
102        import json
103
104        if dep.required_fields:
105            # 從 structured_data 中提取指定的元資料欄位
106            extracted = {}
107            for field in dep.required_fields:
108                if field in source.structured_data:
109                    extracted[field] = source.structured_data[field]
110            data = extracted
111        else:
112            data = source.metadata
113
114        content = f"[{source.agent_name} 元資料]\n{json.dumps(data, ensure_ascii=False, indent=2)}"
115        return self._truncate(content, dep.max_tokens)
116
117    def _extract_full_output(self, source: AgentOutput, dep: ContextDependency) -> str:
118        """提取完整輸出"""
119        content = source.to_full_context()
120        return self._truncate(content, dep.max_tokens)
121
122    def _truncate(self, content: str, max_tokens: Optional[int]) -> str:
123        """截斷內容"""
124        if max_tokens is None:
125            return content
126        # 粗略估算：1 token ≈ 4 字元（中英混合）
127        max_chars = max_tokens * 3
128        if len(content) > max_chars:
129            return content[:max_chars] + "\n...[已截斷]"
130        return content
131
132    def extract_with_llm(
133        self,
134        source: AgentOutput,
135        target_agent: str,
136        extraction_prompt: str
137    ) -> str:
138        """
139        使用 LLM 進行智能提取
140
141        適用於需要更精確理解和提取的情況
142        """
143        prompt = f"""從以下 Agent 輸出中提取 {target_agent} 所需的資訊。
144
145來源 Agent：{source.agent_name}
146來源輸出：
147{source.to_full_context()}
148
149提取要求：
150{extraction_prompt}
151
152請只輸出提取後的內容，保持簡潔。"""
153
154        response = client.messages.create(
155            model="claude-3-5-haiku-20241022",  # 使用便宜模型做提取
156            max_tokens=1000,
157            messages=[{"role": "user", "content": prompt}]
158        )
159
160        return response.content[0].text
161
162
163class ContextBuilder:
164    """
165    Context 建構器
166
167    根據依賴圖為目標 Agent 建構完整的 Context
168    """
169
170    def __init__(
171        self,
172        dependency_graph: dict[str, AgentContextSpec],
173        extractor: Optional[ContextExtractor] = None
174    ):
175        self.dependency_graph = dependency_graph
176        self.extractor = extractor or ContextExtractor()
177        self._output_store: dict[str, AgentOutput] = {}
178
179    def store_output(self, output: AgentOutput):
180        """儲存 Agent 輸出"""
181        self._output_store[output.agent_name] = output
182
183    def build_context(
184        self,
185        target_agent: str,
186        original_task: str
187    ) -> str:
188        """
189        為目標 Agent 建構 Context
190
191        Args:
192            target_agent: 目標 Agent 名稱
193            original_task: 原始任務描述
194
195        Returns:
196            建構好的 Context 字串
197        """
198        spec = self.dependency_graph.get(target_agent)
199        if not spec:
200            return original_task
201
202        context_parts = []
203        total_tokens = 0
204
205        # 1. 添加原始任務（如果需要）
206        if spec.requires_original_task:
207            task_context = f"[原始任務]\n{original_task}"
208            context_parts.append(task_context)
209            total_tokens += len(task_context) // 3  # 粗略估算
210
211        # 2. 按優先級排序依賴
212        sorted_deps = sorted(spec.dependencies, key=lambda d: d.priority, reverse=True)
213
214        # 3. 提取每個依賴的 Context
215        for dep in sorted_deps:
216            if dep.source_agent not in self._output_store:
217                continue
218
219            source_output = self._output_store[dep.source_agent]
220
221            request = ExtractionRequest(
222                source_output=source_output,
223                target_agent=target_agent,
224                dependency=dep
225            )
226
227            extracted = self.extractor.extract(request)
228
229            # 檢查是否會超過總限制
230            extracted_tokens = len(extracted) // 3
231            if total_tokens + extracted_tokens > spec.max_total_context:
232                # 嘗試截斷
233                remaining = spec.max_total_context - total_tokens
234                if remaining > 100:
235                    extracted = extracted[:remaining * 3] + "\n...[已截斷]"
236                else:
237                    continue  # 跳過這個依賴
238
239            context_parts.append(extracted)
240            total_tokens += extracted_tokens
241
242        return "\n\n---\n\n".join(context_parts)
243
244    def get_context_stats(self, target_agent: str) -> dict:
245        """取得 Context 統計"""
246        spec = self.dependency_graph.get(target_agent)
247        if not spec:
248            return {}
249
250        stats = {
251            "target_agent": target_agent,
252            "max_total_context": spec.max_total_context,
253            "dependencies": []
254        }
255
256        for dep in spec.dependencies:
257            dep_info = {
258                "source": dep.source_agent,
259                "type": dep.context_type.value,
260                "max_tokens": dep.max_tokens,
261                "available": dep.source_agent in self._output_store
262            }
263            stats["dependencies"].append(dep_info)
264
265        return stats

  1import anthropic
  2import json
  3from dataclasses import dataclass
  4from typing import Optional
  5
  6client = anthropic.Anthropic()
  7
  8@dataclass
  9class RelevanceScore:
 10    """相關性評分"""
 11    content: str
 12    score: float  # 0-1
 13    reason: str
 14    should_include: bool
 15
 16class RelevanceFilter:
 17    """
 18    相關性過濾器
 19
 20    使用 LLM 判斷內容與目標任務的相關性，
 21    只保留高相關性的內容
 22    """
 23
 24    RELEVANCE_PROMPT = """分析以下內容與目標任務的相關性。
 25
 26目標任務：
 27{task}
 28
 29目標 Agent：{target_agent}
 30目標 Agent 職責：{agent_role}
 31
 32待評估內容：
 33{content}
 34
 35請以 JSON 格式評估相關性：
 36{{
 37    "score": 0.0-1.0,
 38    "reason": "簡短說明",
 39    "key_relevant_parts": ["相關部分1", "相關部分2"],
 40    "irrelevant_parts": ["不相關部分1"]
 41}}
 42
 43評分標準：
 44- 1.0: 完全相關，必須包含
 45- 0.7-0.9: 高度相關，建議包含
 46- 0.4-0.6: 部分相關，可選包含
 47- 0.1-0.3: 低相關，通常不需要
 48- 0.0: 完全不相關
 49
 50只返回 JSON。"""
 51
 52    def __init__(
 53        self,
 54        model: str = "claude-3-5-haiku-20241022",
 55        relevance_threshold: float = 0.5
 56    ):
 57        self.model = model
 58        self.relevance_threshold = relevance_threshold
 59        self._cache: dict[str, RelevanceScore] = {}
 60
 61    def _compute_cache_key(self, task: str, content: str) -> str:
 62        import hashlib
 63        combined = f"{task[:100]}|{content[:200]}"
 64        return hashlib.md5(combined.encode()).hexdigest()
 65
 66    def score_relevance(
 67        self,
 68        content: str,
 69        task: str,
 70        target_agent: str,
 71        agent_role: str
 72    ) -> RelevanceScore:
 73        """評估內容相關性"""
 74        # 檢查快取
 75        cache_key = self._compute_cache_key(task, content)
 76        if cache_key in self._cache:
 77            return self._cache[cache_key]
 78
 79        prompt = self.RELEVANCE_PROMPT.format(
 80            task=task,
 81            target_agent=target_agent,
 82            agent_role=agent_role,
 83            content=content[:3000]  # 限制長度
 84        )
 85
 86        response = client.messages.create(
 87            model=self.model,
 88            max_tokens=500,
 89            messages=[{"role": "user", "content": prompt}]
 90        )
 91
 92        try:
 93            text = response.content[0].text
 94            if "```" in text:
 95                text = text.split("```")[1].split("```")[0]
 96                if text.startswith("json"):
 97                    text = text[4:]
 98            data = json.loads(text.strip())
 99
100            result = RelevanceScore(
101                content=content,
102                score=data.get("score", 0.5),
103                reason=data.get("reason", ""),
104                should_include=data.get("score", 0.5) >= self.relevance_threshold
105            )
106        except (json.JSONDecodeError, KeyError):
107            result = RelevanceScore(
108                content=content,
109                score=0.5,
110                reason="評估失敗，使用預設值",
111                should_include=True
112            )
113
114        self._cache[cache_key] = result
115        return result
116
117    def filter_context_parts(
118        self,
119        parts: list[str],
120        task: str,
121        target_agent: str,
122        agent_role: str,
123        max_parts: Optional[int] = None
124    ) -> list[str]:
125        """
126        過濾 Context 部分，只保留相關內容
127
128        Args:
129            parts: Context 片段列表
130            task: 目標任務
131            target_agent: 目標 Agent
132            agent_role: Agent 職責描述
133            max_parts: 最多保留幾個部分
134
135        Returns:
136            過濾後的 Context 片段列表
137        """
138        scored_parts = []
139
140        for part in parts:
141            if not part.strip():
142                continue
143            score = self.score_relevance(part, task, target_agent, agent_role)
144            if score.should_include:
145                scored_parts.append((score.score, part))
146
147        # 按相關性排序
148        scored_parts.sort(key=lambda x: x[0], reverse=True)
149
150        # 限制數量
151        if max_parts:
152            scored_parts = scored_parts[:max_parts]
153
154        return [part for _, part in scored_parts]
155
156
157class SmartContextFilter:
158    """
159    智能 Context 過濾器
160
161    結合規則和 LLM 的混合過濾策略
162    """
163
164    # 不同 Agent 角色的關鍵字
165    AGENT_KEYWORDS = {
166        "developer": ["程式", "code", "實作", "implement", "API", "函數", "class"],
167        "architect": ["架構", "architecture", "設計", "design", "模式", "pattern"],
168        "reviewer": ["審查", "review", "問題", "issue", "建議", "suggestion"],
169        "doc_writer": ["文件", "document", "說明", "description", "API", "用法"],
170        "tester": ["測試", "test", "用例", "case", "驗證", "verify"]
171    }
172
173    def __init__(
174        self,
175        use_llm_filter: bool = True,
176        relevance_threshold: float = 0.5
177    ):
178        self.use_llm_filter = use_llm_filter
179        self.llm_filter = RelevanceFilter(relevance_threshold=relevance_threshold) if use_llm_filter else None
180
181    def _keyword_filter(self, content: str, target_agent: str) -> bool:
182        """基於關鍵字的快速過濾"""
183        keywords = self.AGENT_KEYWORDS.get(target_agent, [])
184        if not keywords:
185            return True
186        content_lower = content.lower()
187        return any(kw.lower() in content_lower for kw in keywords)
188
189    def filter(
190        self,
191        context_parts: list[str],
192        task: str,
193        target_agent: str,
194        agent_role: str
195    ) -> list[str]:
196        """
197        智能過濾 Context
198
199        流程：
200        1. 關鍵字快速過濾
201        2. LLM 相關性評估（如果啟用）
202        """
203        # 階段 1：關鍵字快速過濾
204        keyword_filtered = [
205            part for part in context_parts
206            if self._keyword_filter(part, target_agent)
207        ]
208
209        # 如果過濾後內容很少，保留全部
210        if len(keyword_filtered) <= 2:
211            keyword_filtered = context_parts
212
213        # 階段 2：LLM 相關性過濾
214        if self.use_llm_filter and self.llm_filter:
215            return self.llm_filter.filter_context_parts(
216                keyword_filtered,
217                task,
218                target_agent,
219                agent_role,
220                max_parts=5
221            )
222
223        return keyword_filtered

  1import anthropic
  2import json
  3from dataclasses import dataclass, field
  4from typing import Optional, Callable
  5from enum import Enum
  6import time
  7
  8client = anthropic.Anthropic()
  9
 10@dataclass
 11class AgentConfig:
 12    """Agent 配置"""
 13    name: str
 14    role: str
 15    system_prompt: str
 16    output_class: type = AgentOutput
 17    model: str = "claude-sonnet-4-20250514"
 18    max_output_tokens: int = 4096
 19
 20@dataclass
 21class TaskResult:
 22    """任務執行結果"""
 23    task_id: str
 24    agent_name: str
 25    output: AgentOutput
 26    context_tokens_used: int
 27    output_tokens: int
 28    execution_time_ms: float
 29    success: bool
 30    error: Optional[str] = None
 31
 32class SelectiveContextOrchestrator:
 33    """
 34    選擇性 Context 傳遞的 Multi-Agent Orchestrator
 35
 36    核心功能：
 37    1. 管理 Agent 依賴關係
 38    2. 建構最小化 Context
 39    3. 協調 Agent 執行
 40    4. 追蹤 Token 使用
 41    """
 42
 43    def __init__(
 44        self,
 45        dependency_graph: dict[str, AgentContextSpec],
 46        enable_relevance_filter: bool = True,
 47        enable_smart_extraction: bool = True
 48    ):
 49        self.dependency_graph = dependency_graph
 50
 51        # 核心組件
 52        self.context_builder = ContextBuilder(dependency_graph)
 53        self.extractor = ContextExtractor(use_llm_extraction=enable_smart_extraction)
 54        self.relevance_filter = SmartContextFilter(use_llm_filter=enable_relevance_filter)
 55
 56        # Agent 註冊表
 57        self.agents: dict[str, AgentConfig] = {}
 58
 59        # 執行記錄
 60        self._execution_history: list[TaskResult] = []
 61        self._stats = {
 62            "total_context_tokens": 0,
 63            "total_output_tokens": 0,
 64            "context_tokens_saved": 0,
 65            "agents_executed": 0
 66        }
 67
 68    def register_agent(self, config: AgentConfig):
 69        """註冊 Agent"""
 70        self.agents[config.name] = config
 71
 72    def _create_agent_executor(self, config: AgentConfig) -> StructuredOutputAgent:
 73        """創建 Agent 執行器"""
 74        return StructuredOutputAgent(
 75            agent_name=config.name,
 76            system_prompt=config.system_prompt,
 77            output_class=config.output_class,
 78            model=config.model
 79        )
 80
 81    def _estimate_tokens(self, text: str) -> int:
 82        """估算 token 數"""
 83        return len(text) // 3  # 粗略估算
 84
 85    def _build_optimized_context(
 86        self,
 87        target_agent: str,
 88        original_task: str
 89    ) -> tuple[str, int, int]:
 90        """
 91        建構優化的 Context
 92
 93        Returns:
 94            (context, actual_tokens, potential_tokens_without_optimization)
 95        """
 96        spec = self.dependency_graph.get(target_agent)
 97        agent_config = self.agents.get(target_agent)
 98
 99        if not spec or not agent_config:
100            return original_task, self._estimate_tokens(original_task), self._estimate_tokens(original_task)
101
102        # 收集所有可能的 Context
103        all_context_parts = []
104        potential_tokens = 0
105
106        # 原始任務
107        if spec.requires_original_task:
108            all_context_parts.append(f"[原始任務]\n{original_task}")
109
110        # 從依賴中提取
111        for dep in spec.dependencies:
112            if dep.source_agent not in self.context_builder._output_store:
113                continue
114
115            source_output = self.context_builder._output_store[dep.source_agent]
116
117            # 計算未優化情況下的 tokens（完整輸出）
118            full_context = source_output.to_full_context()
119            potential_tokens += self._estimate_tokens(full_context)
120
121            # 提取優化後的 Context
122            request = ExtractionRequest(
123                source_output=source_output,
124                target_agent=target_agent,
125                dependency=dep
126            )
127            extracted = self.extractor.extract(request)
128            all_context_parts.append(extracted)
129
130        # 相關性過濾
131        filtered_parts = self.relevance_filter.filter(
132            all_context_parts,
133            original_task,
134            target_agent,
135            agent_config.role
136        )
137
138        # 組合最終 Context
139        final_context = "\n\n---\n\n".join(filtered_parts)
140
141        # 確保不超過限制
142        max_tokens = spec.max_total_context
143        actual_tokens = self._estimate_tokens(final_context)
144
145        if actual_tokens > max_tokens:
146            # 截斷
147            final_context = final_context[:max_tokens * 3] + "\n...[Context 已截斷]"
148            actual_tokens = max_tokens
149
150        return final_context, actual_tokens, potential_tokens
151
152    def execute_agent(
153        self,
154        agent_name: str,
155        original_task: str,
156        task_id: str = ""
157    ) -> TaskResult:
158        """
159        執行單個 Agent
160
161        自動建構最小化 Context 並執行
162        """
163        start_time = time.time()
164
165        agent_config = self.agents.get(agent_name)
166        if not agent_config:
167            return TaskResult(
168                task_id=task_id,
169                agent_name=agent_name,
170                output=AgentOutput(agent_name=agent_name, task_id=task_id, summary=""),
171                context_tokens_used=0,
172                output_tokens=0,
173                execution_time_ms=0,
174                success=False,
175                error=f"Agent {agent_name} 未註冊"
176            )
177
178        # 建構優化 Context
179        context, actual_tokens, potential_tokens = self._build_optimized_context(
180            agent_name, original_task
181        )
182
183        # 執行 Agent
184        executor = self._create_agent_executor(agent_config)
185
186        try:
187            output = executor.execute(original_task, context)
188            output.task_id = task_id
189
190            # 儲存輸出供下游使用
191            self.context_builder.store_output(output)
192
193            execution_time = (time.time() - start_time) * 1000
194            output_tokens = output.metadata.get("output_tokens", 0)
195
196            # 更新統計
197            self._stats["total_context_tokens"] += actual_tokens
198            self._stats["total_output_tokens"] += output_tokens
199            self._stats["context_tokens_saved"] += (potential_tokens - actual_tokens)
200            self._stats["agents_executed"] += 1
201
202            result = TaskResult(
203                task_id=task_id,
204                agent_name=agent_name,
205                output=output,
206                context_tokens_used=actual_tokens,
207                output_tokens=output_tokens,
208                execution_time_ms=execution_time,
209                success=True
210            )
211
212        except Exception as e:
213            result = TaskResult(
214                task_id=task_id,
215                agent_name=agent_name,
216                output=AgentOutput(agent_name=agent_name, task_id=task_id, summary=""),
217                context_tokens_used=actual_tokens,
218                output_tokens=0,
219                execution_time_ms=(time.time() - start_time) * 1000,
220                success=False,
221                error=str(e)
222            )
223
224        self._execution_history.append(result)
225        return result
226
227    def execute_pipeline(
228        self,
229        original_task: str,
230        agent_sequence: list[str],
231        task_id: str = ""
232    ) -> list[TaskResult]:
233        """
234        執行 Agent 管線
235
236        按順序執行多個 Agent，自動傳遞 Context
237        """
238        results = []
239
240        for i, agent_name in enumerate(agent_sequence):
241            print(f"\n{'='*50}")
242            print(f"執行 Agent {i+1}/{len(agent_sequence)}: {agent_name}")
243            print(f"{'='*50}")
244
245            result = self.execute_agent(
246                agent_name,
247                original_task,
248                task_id=f"{task_id}_{i}" if task_id else str(i)
249            )
250
251            results.append(result)
252
253            if result.success:
254                print(f"✅ 成功")
255                print(f"   Context tokens: {result.context_tokens_used}")
256                print(f"   Output tokens: {result.output_tokens}")
257                print(f"   執行時間: {result.execution_time_ms:.0f}ms")
258            else:
259                print(f"❌ 失敗: {result.error}")
260
261        return results
262
263    def get_stats(self) -> dict:
264        """取得統計資訊"""
265        total_potential = self._stats["total_context_tokens"] + self._stats["context_tokens_saved"]
266        savings_pct = (
267            self._stats["context_tokens_saved"] / total_potential * 100
268            if total_potential > 0 else 0
269        )
270
271        return {
272            **self._stats,
273            "total_tokens": self._stats["total_context_tokens"] + self._stats["total_output_tokens"],
274            "context_savings_pct": f"{savings_pct:.1f}%",
275            "execution_count": len(self._execution_history)
276        }
277
278    def get_execution_summary(self) -> str:
279        """取得執行摘要"""
280        lines = ["執行摘要：", ""]
281
282        for result in self._execution_history:
283            status = "✅" if result.success else "❌"
284            lines.append(
285                f"{status} {result.agent_name}: "
286                f"Context {result.context_tokens_used} tok, "
287                f"Output {result.output_tokens} tok, "
288                f"{result.execution_time_ms:.0f}ms"
289            )
290
291        lines.append("")
292        stats = self.get_stats()
293        lines.append(f"總 Context tokens: {stats['total_context_tokens']}")
294        lines.append(f"總 Output tokens: {stats['total_output_tokens']}")
295        lines.append(f"Context 節省: {stats['context_savings_pct']}")
296
297        return "\n".join(lines)
298
299
300# 使用範例
301def create_development_pipeline() -> SelectiveContextOrchestrator:
302    """創建軟體開發管線"""
303
304    orchestrator = SelectiveContextOrchestrator(
305        dependency_graph=CONTEXT_DEPENDENCY_GRAPH,
306        enable_relevance_filter=True,
307        enable_smart_extraction=True
308    )
309
310    # 註冊 Agents
311    orchestrator.register_agent(AgentConfig(
312        name="requirements_analyst",
313        role="需求分析師，負責理解和整理使用者需求",
314        system_prompt="""你是專業的需求分析師。
315分析使用者需求，產出結構化的需求文件。
316重點關注：功能需求、非功能需求、限制條件、使用者故事。""",
317        output_class=RequirementsAnalystOutput
318    ))
319
320    orchestrator.register_agent(AgentConfig(
321        name="architect",
322        role="系統架構師，負責設計系統架構和技術選型",
323        system_prompt="""你是資深系統架構師。
324根據需求設計系統架構，包括技術選型、API 設計、資料模型。
325考慮可擴展性、可維護性、效能需求。""",
326        output_class=ArchitectOutput
327    ))
328
329    orchestrator.register_agent(AgentConfig(
330        name="developer",
331        role="軟體開發者，負責實作程式碼",
332        system_prompt="""你是資深軟體開發者。
333根據架構設計實作乾淨、可維護的程式碼。
334遵循最佳實踐和程式碼規範。""",
335        output_class=DeveloperOutput
336    ))
337
338    orchestrator.register_agent(AgentConfig(
339        name="code_reviewer",
340        role="程式碼審查者，負責審查程式碼品質",
341        system_prompt="""你是程式碼審查專家。
342審查程式碼的品質、安全性、效能和可維護性。
343提供具體的改進建議。""",
344        output_class=CodeReviewerOutput,
345        model="claude-3-5-haiku-20241022"  # 審查用較輕量模型
346    ))
347
348    orchestrator.register_agent(AgentConfig(
349        name="doc_writer",
350        role="技術文件撰寫者，負責撰寫文件",
351        system_prompt="""你是技術文件撰寫專家。
352撰寫清晰、完整的技術文件。
353包括 API 文件、使用說明、架構說明。""",
354        model="claude-3-5-haiku-20241022"
355    ))
356
357    return orchestrator
358
359
360if __name__ == "__main__":
361    # 創建管線
362    orchestrator = create_development_pipeline()
363
364    # 執行任務
365    task = """
366    開發一個簡單的待辦事項 API，需要：
367    1. CRUD 操作（建立、讀取、更新、刪除待辦事項）
368    2. 使用者認證（JWT）
369    3. 待辦事項可以設定優先級和截止日期
370    4. 支援分頁查詢
371    """
372
373    results = orchestrator.execute_pipeline(
374        original_task=task,
375        agent_sequence=[
376            "requirements_analyst",
377            "architect",
378            "developer",
379            "code_reviewer",
380            "doc_writer"
381        ],
382        task_id="todo_api"
383    )
384
385    print("\n" + "="*60)
386    print(orchestrator.get_execution_summary())