# 使用 Anthropic Python SDK 定義工具
 
from anthropic import Anthropic
 
client = Anthropic()
 
tools = [
    {
        "name": "query_customer",
        "description": "按名稱或統編查詢客戶資訊（包括聯絡方式、信用額度、歷史交易）",
        "input_schema": {
            "type": "object",
            "properties": {
                "identifier": {
                    "type": "string",
                    "description": "客戶名稱或統一編號"
                }
            },
            "required": ["identifier"]
        }
    },
    {
        "name": "create_invoice",
        "description": "建立新發票，自動計算稅金與運費",
        "input_schema": {
            "type": "object",
            "properties": {
                "customer_id": {"type": "string"},
                "items": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "product_id": {"type": "string"},
                            "quantity": {"type": "number"}
                        }
                    }
                },
                "due_date": {
                    "type": "string",
                    "description": "繳款期限，格式 YYYY-MM-DD"
                }
            },
            "required": ["customer_id", "items"]
        }
    },
    {
        "name": "send_email",
        "description": "發送電子郵件，支援附件",
        "input_schema": {
            "type": "object",
            "properties": {
                "to": {"type": "string"},
                "subject": {"type": "string"},
                "body": {"type": "string"},
                "attachments": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "檔案路徑陣列"
                }
            },
            "required": ["to", "subject", "body"]
        }
    }
]
 
# Agent 執行迴圈
messages = [
    {"role": "user", "content": "幫我為客戶 ABC 開一張 50,000 元的發票並寄出去"}
]
 
response = client.messages.create(
    model="claude-sonnet-4-6",  # 當前 Claude Sonnet 4.6（速度與智能的最佳平衡）
    max_tokens=4096,
    tools=tools,
    messages=messages
)
 
# 處理 Agent 的工具呼叫
while response.stop_reason == "tool_use":
    tool_use = next(
        block for block in response.content
        if block.type == "tool_use"
    )
 
    # 這裡實現工具的實際邏輯
    if tool_use.name == "query_customer":
        result = external_api.query_customer(tool_use.input["identifier"])
    elif tool_use.name == "create_invoice":
        result = external_api.create_invoice(**tool_use.input)
    elif tool_use.name == "send_email":
        result = external_api.send_email(**tool_use.input)
 
    # 將結果送回 Agent
    messages.append({"role": "assistant", "content": response.content})
    messages.append({
        "role": "user",
        "content": [
            {
                "type": "tool_result",
                "tool_use_id": tool_use.id,
                "content": str(result)
            }
        ]
    })
 
    response = client.messages.create(
        model="claude-sonnet-4-6",  # 當前 Claude Sonnet 4.6（速度與智能的最佳平衡）
        max_tokens=4096,
        tools=tools,
        messages=messages
    )
 
# 最終回應
final_response = next(
    (block.text for block in response.content if hasattr(block, "text")),
    None
)
print(final_response)

# 範例：儲存過去成功的推薦案例
from pinecone import Pinecone
 
pc = Pinecone(api_key="xxx")
index = pc.Index("customer-insights")
 
# 執行推薦後，儲存案例
embedding = get_embedding(
    "製造業、年營收 1000 萬、購買過工業軟體 → 推薦 ERP 升級方案"
)
index.upsert([
    ("case_001", embedding, {
        "industry": "manufacturing",
        "revenue_range": "10M-50M",
        "past_purchases": "industrial_software",
        "recommendation": "ERP_upgrade",
        "success_rate": 0.78
    })
])
 
# 下次推薦時查詢相似案例
query_embedding = get_embedding(
    "製造業、年營收 1500 萬、購買過工業軟體"
)
similar_cases = index.query(vector=query_embedding, top_k=5)
# 返回最相似的 5 個過往成功案例，Agent 參考做決策

class Agent:
    def __init__(self, name, system_prompt, tools):
        self.name = name
        self.system_prompt = system_prompt
        self.tools = tools
        self.conversation_history = []
        self.vector_store = None
        self.working_memory = {}
 
    def execute(self, user_message, max_iterations=10):
        """單一 Agent 的執行迴圈"""
        self.conversation_history.append({
            "role": "user",
            "content": user_message
        })
 
        for iteration in range(max_iterations):
            # 1. 呼叫 LLM 進行推理
            response = self.call_llm()
 
            # 2. 檢查是否需要呼叫工具
            if response.stop_reason == "tool_use":
                tool_calls = [b for b in response.content if b.type == "tool_use"]
 
                # 3. 執行工具（可並行）
                results = self.execute_tools(tool_calls)
 
                # 4. 將結果反饋給 LLM
                self.conversation_history.append({
                    "role": "assistant",
                    "content": response.content
                })
                self.conversation_history.append({
                    "role": "user",
                    "content": results
                })
            else:
                # Agent 完成，回傳最終答案
                return next((b.text for b in response.content if hasattr(b, "text")), None)
 
        # 超過最大迭代次數
        raise Exception("Agent 未能在規定迴圈內完成任務")

class ControllableAgent:
    def __init__(self, approval_threshold="critical"):
        self.approval_threshold = approval_threshold  # "none", "critical", "all"
 
    def execute_action(self, action):
        """根據設定的門檻進行人工審批"""
        action_risk_level = self.assess_risk(action)
 
        # 風險分級
        if action_risk_level == "low" and self.approval_threshold == "none":
            # 低風險 + 無需審批：直接執行
            return self.execute(action)
 
        elif action_risk_level == "medium" and self.approval_threshold == "none":
            # 中等風險 + 無需審批：直接執行（但記錄）
            self.audit_log(action)
            return self.execute(action)
 
        elif action_risk_level == "critical" or self.approval_threshold == "critical":
            # 高風險或設定為「重要審批」：等待人類確認
            approval_token = self.request_approval(action, timeout=3600)  # 1 小時超時
            if approval_token.approved:
                return self.execute(action)
            else:
                return {"status": "rejected", "reason": approval_token.reason}
 
        elif self.approval_threshold == "all":
            # 全部審批：所有動作都需確認
            approval_token = self.request_approval(action, timeout=300)
            if approval_token.approved:
                return self.execute(action)
            else:
                return {"status": "rejected"}
 
    def assess_risk(self, action):
        """評估動作風險等級"""
        if action.type == "delete_customer":
            return "critical"  # 刪除客戶 → 高風險
        elif action.type == "refund_over_10k":
            return "critical"  # 退款超過 10K → 高風險
        elif action.type == "send_email":
            return "low"       # 發信 → 低風險
        elif action.type == "modify_invoice":
            return "medium"    # 修改發票 → 中等風險
        else:
            return "low"
 
    def request_approval(self, action, timeout=3600):
        """向人類發送審批請求"""
        notification = {
            "type": "approval_request",
            "action": action.describe(),
            "risk_level": self.assess_risk(action),
            "timeout_seconds": timeout,
            "approval_link": f"https://company.com/approvals/{uuid.uuid4()}"
        }
        # 發送至信箱、Slack、內部系統等
        self.send_notification(notification)
 
        # 等待結果（同步阻塞或非同步 callback）
        return self.wait_for_approval(timeout)

import json
import logging
from datetime import datetime
from enum import Enum
 
class DecisionLog:
    def __init__(self, db_connection):
        self.db = db_connection
        self.logger = logging.getLogger("agent_audit")
 
    def log_reasoning_step(self, agent_id, step_number, thought, action, observation):
        """記錄 ReAct 迴圈的每一步"""
        record = {
            "timestamp": datetime.utcnow().isoformat(),
            "agent_id": agent_id,
            "step": step_number,
            "thinking": thought,
            "action_taken": action,
            "observation_result": observation,
            "tokens_used": thought.token_count + action.token_count
        }
        self.db.audit_logs.insert_one(record)
        self.logger.info(f"Agent {agent_id} Step {step_number}: {action.type}")
 
    def log_tool_call(self, agent_id, tool_name, input_params, output, execution_time_ms):
        """記錄工具呼叫及結果"""
        record = {
            "timestamp": datetime.utcnow().isoformat(),
            "agent_id": agent_id,
            "tool_name": tool_name,
            "input_params": self.sanitize(input_params),  # 移除敏感資訊
            "output_summary": self.summarize(output),
            "execution_time_ms": execution_time_ms,
            "status": "success" if output.error is None else "failure",
            "error_message": output.error if output.error else None
        }
        self.db.tool_calls.insert_one(record)
 
    def log_decision_rationale(self, agent_id, decision, evidence, confidence_score):
        """記錄決策理由與信心度"""
        record = {
            "timestamp": datetime.utcnow().isoformat(),
            "agent_id": agent_id,
            "decision": decision,
            "supporting_evidence": evidence,  # 引用的資訊來源
            "confidence_score": confidence_score,  # 0.0 - 1.0
            "reversible": decision.is_reversible
        }
        self.db.decisions.insert_one(record)
 
    def generate_audit_report(self, task_id, format="html"):
        """生成任務的完整審計報告"""
        logs = self.db.audit_logs.find({"task_id": task_id})
        tool_calls = self.db.tool_calls.find({"task_id": task_id})
        decisions = self.db.decisions.find({"task_id": task_id})
 
        report = {
            "task_id": task_id,
            "total_reasoning_steps": len(logs),
            "tools_used": len(tool_calls),
            "decisions_made": len(decisions),
            "total_tokens": sum(log["tokens_used"] for log in logs),
            "timeline": self.build_timeline(logs, tool_calls, decisions)
        }
 
        if format == "html":
            return self.render_html_report(report)
        elif format == "json":
            return json.dumps(report)
        else:
            return report
 
# 使用範例
audit = DecisionLog(mongodb_client)
 
# Agent 每次思考/行動都記錄
audit.log_reasoning_step(
    agent_id="agent_sales_001",
    step_number=3,
    thought="客戶 ABC 已購買過 3 次 ERP 軟體，應該推薦進階版本或追加模組",
    action={"type": "tool_call", "tool": "recommend_products", "params": {"customer_id": 12345}},
    observation={"products": ["ERP-advanced", "HR-module"], "confidence": 0.92}
)
 
# 外部系統呼叫也記錄
audit.log_tool_call(
    agent_id="agent_sales_001",
    tool_name="query_customer_history",
    input_params={"customer_id": 12345},
    output={"purchases": [...]},
    execution_time_ms=245
)
 
# 最後生成稽核報告（供合規部門檢查）
report = audit.generate_audit_report(task_id="task_proposal_2026_03_26", format="html")

from tenacity import retry, stop_after_attempt, wait_exponential
from circuitbreaker import circuit
 
class ResilientAgent:
    def __init__(self):
        self.circuit_breaker_states = {}  # 追蹤各工具的健康狀態
 
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10)
    )
    def call_external_api(self, api_name, params):
        """帶有自動重試的 API 呼叫"""
        try:
            response = api_client.call(api_name, params)
            return response
        except TimeoutError:
            # 重試邏輯由 @retry 裝飾器自動處理
            raise
        except Exception as e:
            self.logger.error(f"API {api_name} 失敗：{e}")
            raise
 
    @circuit(failure_threshold=5, recovery_timeout=60)
    def query_customer_database(self, customer_id):
        """使用熔斷器保護資料庫連線"""
        return db.customers.find_one({"id": customer_id})
 
    def handle_tool_failure(self, tool_name, error_type, original_params):
        """優雅降級策略"""
        if error_type == "timeout":
            # 對於超時，嘗試使用快取資料
            cached = self.get_cached_result(tool_name, original_params)
            if cached:
                self.log_fallback("used_cache", tool_name)
                return {"status": "degraded", "source": "cache", "data": cached}
            else:
                self.log_fallback("no_cache", tool_name)
                return {"status": "error", "reason": "無法獲取數據，已超時"}
 
        elif error_type == "circuit_open":
            # 熔斷器開啟（服務故障），使用離線替代方案
            if tool_name == "query_competitor_pricing":
                # 查詢 3 天前的快取價格
                return self.get_stale_data(tool_name, max_age_hours=72)
            else:
                return {"status": "error", "reason": f"{tool_name} 暫時不可用，請稍後重試"}
 
        elif error_type == "permission_denied":
            # 權限不足，提示使用者而非重試
            return {"status": "error", "reason": "權限不足，無法存取該資源"}
 
        else:
            # 其他錯誤
            return {"status": "error", "reason": str(error_type)}
 
    def execute_with_fallback(self, primary_tool, fallback_tools, params):
        """優先嘗試主要工具，失敗則依序嘗試備選工具"""
        tools_to_try = [primary_tool] + fallback_tools
        last_error = None
 
        for tool in tools_to_try:
            try:
                self.logger.info(f"嘗試工具: {tool}")
                result = self.call_tool(tool, params)
 
                if result.status == "success":
                    self.logger.info(f"成功使用工具: {tool}")
                    return result
                else:
                    last_error = result
 
            except Exception as e:
                self.logger.warning(f"工具 {tool} 失敗: {e}")
                last_error = e
                continue
 
        # 所有工具都失敗
        self.logger.error(f"所有工具均失敗: {last_error}")
        return {"status": "error", "reason": "無可用工具", "details": str(last_error)}
 
# 使用範例
agent = ResilientAgent()
 
# 優先使用即時 API，失敗則使用快取，再失敗則使用統計預測
result = agent.execute_with_fallback(
    primary_tool="get_live_competitor_pricing",
    fallback_tools=["get_cached_pricing", "estimate_based_on_history"],
    params={"competitor_id": "competitor_001"}
)

class SecureToolRegistry:
    def __init__(self):
        self.tool_permissions = {
            "sales_agent": {
                "query_customer": {"allowed": True, "fields": ["name", "contact", "purchase_history"]},
                "query_customer.salary": {"allowed": False},  # 銷售員不該看薪水
                "create_invoice": {"allowed": True, "max_amount": 500000},  # 最多開 50 萬發票
                "delete_customer": {"allowed": False},
                "send_email": {"allowed": True, "allowed_domains": ["@company.tw", "@partner.com"]}
            },
            "hr_agent": {
                "query_employee": {"allowed": True, "fields": ["name", "dept", "salary", "performance"]},
                "query_customer": {"allowed": False},  # HR Agent 不該看客戶資料
                "send_email": {"allowed": True, "allowed_domains": ["@company.tw"]}
            },
            "analyst_agent": {
                "query_customer": {"allowed": True, "fields": ["*"]},  # 分析員可看全部
                "generate_report": {"allowed": True},
                "send_email": {"allowed": False}  # 分析員不該直接發信
            }
        }
 
    def check_permission(self, agent_role, tool_name, params):
        """檢查 Agent 是否有權限呼叫該工具"""
        if agent_role not in self.tool_permissions:
            raise PermissionError(f"未知的 Agent 角色: {agent_role}")
 
        if tool_name not in self.tool_permissions[agent_role]:
            raise PermissionError(f"Agent {agent_role} 無權使用工具 {tool_name}")
 
        permission = self.tool_permissions[agent_role][tool_name]
 
        if not permission.get("allowed", False):
            raise PermissionError(f"Agent {agent_role} 被禁止使用工具 {tool_name}")
 
        # 檢查參數限制
        if "max_amount" in permission:
            if params.get("amount", 0) > permission["max_amount"]:
                raise PermissionError(
                    f"金額超過限制: {params['amount']} > {permission['max_amount']}"
                )
 
        if "fields" in permission and permission["fields"] != ["*"]:
            # 限制可查詢的欄位
            params["allowed_fields"] = permission["fields"]
 
        return True
 
class SensitiveDataMasker:
    """對敏感資訊進行遮蔽"""
 
    def mask_result(self, tool_name, result, agent_role):
        """根據 Agent 角色遮蔽結果中的敏感欄位"""
        if tool_name == "query_customer":
            if agent_role == "sales_agent":
                # 銷售員看不到稅籍編號、銀行帳戶
                result.pop("tax_id", None)
                result.pop("bank_account", None)
            elif agent_role == "analyst_agent":
                # 分析員可以看全部
                pass
 
        elif tool_name == "query_employee":
            if agent_role == "hr_agent":
                # HR 員工看得到薪水，但要遮蔽其他敏感欄位
                result.pop("ssn", None)  # 身分證號
                result.pop("home_address", None)
 
        return result
 
# 使用範例
registry = SecureToolRegistry()
 
# Sales Agent 嘗試查詢客戶
try:
    registry.check_permission(
        agent_role="sales_agent",
        tool_name="query_customer",
        params={"customer_id": 12345}
    )
    result = execute_tool("query_customer", customer_id=12345)
    result = SensitiveDataMasker().mask_result("query_customer", result, "sales_agent")
except PermissionError as e:
    print(f"拒絕: {e}")
 
# Sales Agent 嘗試刪除客戶 → 被拒絕
try:
    registry.check_permission(
        agent_role="sales_agent",
        tool_name="delete_customer",
        params={"customer_id": 12345}
    )
except PermissionError as e:
    print(f"拒絕: {e}")  # "Agent sales_agent 被禁止使用工具 delete_customer"
 
# Sales Agent 嘗試開超過額度的發票 → 被拒絕
try:
    registry.check_permission(
        agent_role="sales_agent",
        tool_name="create_invoice",
        params={"amount": 600000}  # 超過 50 萬額度
    )
except PermissionError as e:
    print(f"拒絕: {e}")  # "金額超過限制: 600000 > 500000"

class FeedbackCollector:
    """收集用戶與系統對 Agent 決策的評價"""
 
    def collect_user_feedback(self, task_id, user_rating, comment=""):
        """用戶評分：Agent 的建議有沒有幫助？"""
        record = {
            "task_id": task_id,
            "feedback_type": "user_rating",
            "rating": user_rating,  # 1-5 stars
            "comment": comment,
            "timestamp": datetime.utcnow().isoformat()
        }
        self.db.feedback.insert_one(record)
 
    def collect_outcome_feedback(self, task_id, metric_name, actual_value, target_value):
        """結果反饋：Agent 的決策帶來什麼實際效果？"""
        record = {
            "task_id": task_id,
            "feedback_type": "outcome",
            "metric": metric_name,
            "actual": actual_value,
            "target": target_value,
            "success": actual_value >= target_value,
            "timestamp": datetime.utcnow().isoformat()
        }
        self.db.feedback.insert_one(record)
 
class ABTestFramework:
    """實驗框架：測試不同的 Agent 提示詞、工具組合、決策邏輯"""
 
    def run_ab_test(self, variant_a, variant_b, sample_size=1000, duration_days=7):
        """
        A/B 測試兩個 Agent 變體
 
        variant_a: {"prompt": "...", "tools": [...], "name": "current"}
        variant_b: {"prompt": "...", "tools": [...], "name": "experimental"}
        """
        test_config = {
            "test_id": uuid.uuid4(),
            "variant_a": variant_a,
            "variant_b": variant_b,
            "sample_size": sample_size,
            "duration_days": duration_days,
            "start_time": datetime.utcnow(),
            "status": "running"
        }
 
        self.db.ab_tests.insert_one(test_config)
 
        return {
            "test_id": test_config["test_id"],
            "routing_rule": f"if random() < 0.5 then variant_a else variant_b"
        }
 
    def analyze_ab_test_results(self, test_id):
        """分析 A/B 測試結果"""
        test = self.db.ab_tests.find_one({"test_id": test_id})
 
        variant_a_feedback = list(self.db.feedback.find({
            "test_id": test_id,
            "variant": "a"
        }))
 
        variant_b_feedback = list(self.db.feedback.find({
            "test_id": test_id,
            "variant": "b"
        }))
 
        result = {
            "test_id": test_id,
            "variant_a": {
                "name": test["variant_a"]["name"],
                "avg_rating": sum(f["rating"] for f in variant_a_feedback) / len(variant_a_feedback),
                "success_rate": sum(1 for f in variant_a_feedback if f.get("success")) / len(variant_a_feedback),
                "sample_size": len(variant_a_feedback)
            },
            "variant_b": {
                "name": test["variant_b"]["name"],
                "avg_rating": sum(f["rating"] for f in variant_b_feedback) / len(variant_b_feedback),
                "success_rate": sum(1 for f in variant_b_feedback if f.get("success")) / len(variant_b_feedback),
                "sample_size": len(variant_b_feedback)
            }
        }
 
        # 統計顯著性檢驗
        from scipy.stats import ttest_ind
 
        ratings_a = [f["rating"] for f in variant_a_feedback]
        ratings_b = [f["rating"] for f in variant_b_feedback]
 
        t_stat, p_value = ttest_ind(ratings_a, ratings_b)
        result["statistical_significance"] = {
            "t_statistic": t_stat,
            "p_value": p_value,
            "is_significant_at_005": p_value < 0.05
        }
 
        # 建議
        if result["variant_b"]["avg_rating"] > result["variant_a"]["avg_rating"] and p_value < 0.05:
            result["recommendation"] = "採用 Variant B（有統計顯著差異）"
        else:
            result["recommendation"] = "保持 Variant A（無明顯改進或不具統計顯著性）"
 
        return result
 
# 使用範例
feedback = FeedbackCollector()
ab_test = ABTestFramework()
 
# 1. 部署新版提示詞做 A/B 測試
test_id = ab_test.run_ab_test(
    variant_a={
        "name": "current",
        "prompt": "你是銷售推薦 Agent。推薦產品時應考量客戶規模、產業、過往購買..."
    },
    variant_b={
        "name": "experimental",
        "prompt": "你是銷售推薦 Agent。在推薦前，先詢問客戶的痛點，再基於痛點推薦..."
    },
    sample_size=500,
    duration_days=14
)
 
# 2. 收集用戶反饋
for task in completed_tasks:
    feedback.collect_user_feedback(
        task_id=task["id"],
        user_rating=user_input["rating"],  # 1-5
        comment=user_input["comment"]
    )
 
# 3. 等待 14 天，分析結果
results = ab_test.analyze_ab_test_results(test_id)
print(f"""
Variant A: 平均評分 {results['variant_a']['avg_rating']:.2f}
Variant B: 平均評分 {results['variant_b']['avg_rating']:.2f}
P-value: {results['statistical_significance']['p_value']:.4f}
建議: {results['recommendation']}
""")

# MVP Agent 框架（簡化版）
 
class MinimumViableAgent:
    def __init__(self):
        self.client = Anthropic()
        self.model = "claude-haiku-4-5-20251001"  # 快速、便宜
        self.tools = [
            "query_database",
            "send_email",
            "create_document"
        ]
        self.max_iterations = 5
 
    def run(self, user_request):
        """最小化 Agent 執行邏輯"""
        messages = [{"role": "user", "content": user_request}]
 
        for iteration in range(self.max_iterations):
            response = self.client.messages.create(
                model=self.model,
                max_tokens=1024,
                tools=self.get_tool_definitions(),
                messages=messages
            )
 
            if response.stop_reason == "end_turn":
                # Agent 完成，回傳結果
                return next(
                    (b.text for b in response.content if hasattr(b, "text")),
                    None
                )
 
            elif response.stop_reason == "tool_use":
                # 執行工具
                tool_use = next(b for b in response.content if b.type == "tool_use")
                result = self.execute_tool(tool_use.name, tool_use.input)
 
                messages.append({"role": "assistant", "content": response.content})
                messages.append({
                    "role": "user",
                    "content": [
                        {
                            "type": "tool_result",
                            "tool_use_id": tool_use.id,
                            "content": json.dumps(result)
                        }
                    ]
                })
 
        return "Agent 無法在規定迴圈內完成任務"
 
# 第一週：驗證假設
agent = MinimumViableAgent()
result = agent.run("幫我為客戶 ABC 準備拜訪資料")
print(result)
 
# 收集反饋
user_rating = input("這個結果有幫助嗎？(1-5)")
time_saved = float(input("省了多少分鐘？"))