result = await client.messages.create(
    model="claude-opus-4-20250514",
    messages=[{"role": "user", "content": "帮我查一下北京天气"}],
    tools=[{"name": "get_weather", ...}]
)

# 好
state = AgentState(goal=goal)
while not self._should_terminate(state): ...

# 不好：状态散落在 self 属性里
while self.iterations < self.max_iterations:
    self.messages.append(...)

# 好：先检查再执行
while not self._should_terminate(state): ...

# 不好：先执行再检查
while True:
    response = await self._think(state)
    if self._should_terminate(state): break

                    ┌─────────────┐
                    │  主Agent    │
                    │ (协调者)    │
                    └──────┬──────┘
          ┌────────────────┼────────────────┐
    ┌─────▼─────┐   ┌─────▼─────┐   ┌─────▼─────┐
    │ 子Agent1  │   │ 子Agent2  │   │ 子Agent3  │
    │ (酒店)    │   │ (天气)    │   │ (预算)    │
    └───────────┘   └───────────┘   └───────────┘

class MasterAgent:
    def __init__(self):
        self.sub_agents = {"hotel": HotelAgent(), "weather": WeatherAgent(), "budget": BudgetAgent()}

    async def run(self, task: str) -> dict:
        subtasks = await self._decompose_task(task)
        results = await asyncio.gather(*[
            self._run_subagent(name, subtask) for name, subtask in subtasks.items()
        ])
        return await self._aggregate_results(results)

┌───────────┐    ┌───────────┐    ┌───────────┐    ┌───────────┐
│  Agent1   │ →  │  Agent2   │ →  │  Agent3   │ →  │  Agent4   │
│ (预处理)   │    │ (核心处理) │    │ (验证)    │    │ (输出)    │
└───────────┘    └───────────┘    └───────────┘    └───────────┘

class ChainAgent:
    def __init__(self):
        self.chain = [PreprocessAgent(), CoreAgent(), ValidationAgent(), OutputAgent()]

    async def run(self, input_data: Any) -> Any:
        current = input_data
        for agent in self.chain:
            current = await agent.process(current)
            if not self._is_valid(current):
                return await self._handle_error(current, agent)
        return current

class PeerNetwork:
    """没有中心协调者，Agent 之间可直接通信"""
    async def run(self, task: str) -> dict:
        queue = MessageQueue()
        agents = [Agent(i) for i in range(3)]
        tasks = [agent.run(queue) for agent in agents]
        await queue.publish("task", task)
        results = await asyncio.gather(*tasks)
        return self._merge_results(results)

async def parallel_execution(tasks: list[dict]) -> list[dict]:
    async def execute_single(task: dict) -> dict:
        agent = create_agent(task["type"])
        result = await agent.run(task["input"])
        return {"task_id": task["id"], "result": result}

    return await asyncio.gather(*[execute_single(t) for t in tasks])

# 适合并行：任务之间无依赖
parallel_tasks = [
    {"id": 1, "type": "search", "input": "北京天气"},
    {"id": 2, "type": "search", "input": "上海天气"},
    {"id": 3, "type": "search", "input": "广州天气"},
]

# 不适合并行：存在依赖关系
sequential_tasks = [
    {"id": 1, "type": "search", "input": "用户ID"},
    {"id": 2, "type": "fetch", "input": "依赖任务1的结果"},
    {"id": 3, "type": "save", "input": "依赖任务2的结果"},
]

async def chain_execution(tasks: list[dict]) -> dict:
    context = {}
    for task in tasks:
        task_input = await self._prepare_input(task, context)
        agent = create_agent(task["type"])
        result = await agent.run(task_input)
        context[task["id"]] = result
        if not self._is_success(result):
            if task.get("retry"):
                result = await self._retry(task, context)
            else:
                raise ExecutionError(f"Task {task['id']} failed")
    return context

class HybridExecutor:
    """并行 + 链式结合：先并行执行独立任务，再按依赖链式执行"""
    async def run(self, workflow: dict) -> dict:
        parallel_results = await self._parallel_phase(workflow["parallel_tasks"])
        context = {"parallel": parallel_results}
        chain_result = await self._chain_phase(workflow["chain_tasks"], context)
        return {"parallel_results": parallel_results, "chain_result": chain_result}

    async def _parallel_phase(self, tasks: list[dict]) -> dict:
        results = await asyncio.gather(*[self._execute(task) for task in tasks])
        return {task["id"]: result for task, result in zip(tasks, results)}

    async def _chain_phase(self, tasks: list[dict], context: dict) -> dict:
        for task in tasks:
            context[task["id"]] = await self._execute(task, context)
        return context

class StateManager:
    def __init__(self, max_history: int = 10):
        self.max_history = max_history
        self.summaries = []
        self.working_memory = {}

    def add_interaction(self, user_msg: str, assistant_msg: str):
        self.summaries.append({"user": user_msg, "assistant": assistant_msg, "timestamp": now()})
        if len(self.summaries) > self.max_history:
            self._compress()

    def _compress(self):
        recent = self.summaries[-self.max_history:]
        older = self.summaries[:-self.max_history]
        summary = self.client.messages.create(
            model="claude-opus-4-20250514",
            messages=[{"role": "user", "content": f"请总结以下对话的关键信息：{self._format_conversation(older)}\n提取：1.用户的主要目标 2.已完成的关键步骤 3.当前状态 4.重要的中间结果"}]
        )
        self.working_memory["conversation_summary"] = summary.content[0].text
        self.summaries = recent

    def get_context_for_llm(self) -> str:
        parts = []
        if "conversation_summary" in self.working_memory:
            parts.append(f"对话摘要：{self.working_memory['conversation_summary']}")
        for item in self.summaries[-self.max_history:]:
            parts.append(f"用户：{item['user']}\n助手：{item['assistant']}")
        return "\n".join(parts)

class SmartContextManager:
    def __init__(self, max_tokens: int = 150000):
        self.max_tokens = max_tokens
        self.priority_levels = {
            "critical": ["目标", "关键约束", "核心决策"],
            "important": ["中间结果", "用户偏好", "当前状态"],
            "normal": ["一般对话", "解释说明"],
            "discardable": ["问候", "重复确认"],
        }

    def build_context(self, all_items: list[dict], current_task: str) -> str:
        weights = {"critical": 1.0, "important": 0.7, "normal": 0.4, "discardable": 0.0}
        scored = []
        for item in all_items:
            relevance = self._calculate_relevance(item, current_task)
            priority = self._get_priority(item)
            scored.append((relevance * weights.get(priority, 0.5), item))

        scored.sort(key=lambda x: x[0], reverse=True)
        selected, total = [], 0
        for _, item in scored:
            tokens = self._estimate_tokens(item)
            if total + tokens <= self.max_tokens:
                selected.append(item)
                total += tokens
        return self._format_selected(selected)

class RetryPolicy:
    def __init__(self, max_retries: int = 3, base_delay: float = 1.0,
                 exponential_base: float = 2.0, max_delay: float = 60.0, jitter: bool = True):
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.exponential_base = exponential_base
        self.max_delay = max_delay
        self.jitter = jitter

    def get_delay(self, attempt: int) -> float:
        delay = min(self.base_delay * (self.exponential_base ** attempt), self.max_delay)
        if self.jitter:
            import random
            delay *= 0.5 + random.random()
        return delay

class FallbackManager:
    def __init__(self):
        self.fallback_map = {
            "primary_search": ["fallback_search_1", "fallback_search_2"],
            "primary_translate": ["fallback_translate"],
            "primary_code_exec": ["fallback_sandbox"],
        }
        self.usage_stats = {}

    async def execute_with_fallback(self, primary_tool: str, args: dict) -> Any:
        last_error = None
        for tool in [primary_tool] + self.fallback_map.get(primary_tool, []):
            try:
                result = await self._execute_tool(tool, args)
                self._record_success(tool)
                return result
            except Exception as e:
                last_error = e
                self._record_failure(tool, e)
        raise AllToolsFailedError([primary_tool] + self.fallback_map.get(primary_tool, []), last_error)

from enum import Enum

class PermissionScope(Enum):
    NONE = "none"
    READ = "read"
    WRITE = "write"
    EXECUTE = "execute"
    ADMIN = "admin"

class Permission:
    def __init__(self, scopes: list[PermissionScope] = None, file_paths: list[str] = [],
                 allowed_tools: list[str] = [], allowed_domains: list[str] = [],
                 max_execution_time: int = 300, max_api_calls: int = 100):
        self.scopes = scopes or [PermissionScope.READ]
        self.file_paths = file_paths
        self.allowed_tools = allowed_tools
        self.allowed_domains = allowed_domains
        self.max_execution_time = max_execution_time
        self.max_api_calls = max_api_calls

class SecurityManager:
    def __init__(self, permission: Permission):
        self.permission = permission
        self.audit_log = []

    def check_file_access(self, path: str, mode: str) -> bool:
        if mode == "write" and PermissionScope.WRITE not in self.permission.scopes:
            return False
        import os
        real = os.path.realpath(path)
        return any(real.startswith(os.path.realpath(p)) for p in self.permission.file_paths)

    def check_tool_usage(self, tool_name: str) -> bool:
        return tool_name in self.permission.allowed_tools

    def check_network_access(self, domain: str) -> bool:
        return any(domain.endswith(a) or domain == a for a in self.permission.allowed_domains)

    def audit(self, operation: str, details: dict):
        self.audit_log.append({"timestamp": now(), "operation": operation, "details": details})

class SandboxConfig:
    def __init__(self, use_sandbox: bool = True, network_isolation: bool = True,
                 filesystem_boundary: str = "/workspace/sandbox", memory_limit: str = "2GB"):
        self.use_sandbox = use_sandbox
        self.network_isolation = network_isolation
        self.filesystem_boundary = filesystem_boundary
        self.memory_limit = memory_limit

class SandboxExecutor:
    def __init__(self, config: SandboxConfig):
        self.config = config

    async def execute(self, code: str, language: str) -> Any:
        if not self.config.use_sandbox:
            return await self._execute_direct(code, language)
        sandbox = await self._prepare_sandbox()
        sandbox.set_memory_limit(self.config.memory_limit)
        sandbox.set_network_isolation(self.config.network_isolation)
        sandbox.set_filesystem_boundary(self.config.filesystem_boundary)
        try:
            return await sandbox.run(code, language)
        finally:
            await sandbox.cleanup()

class AgentMonitor:
    metrics = {
        "requests_total": "总请求数", "requests_success": "成功请求数", "requests_failed": "失败请求数",
        "average_latency": "平均延迟", "p99_latency": "P99 延迟", "active_agents": "活跃Agent数",
        "tools_usage": "工具使用统计", "error_distribution": "错误分布",
    }

class AlertManager:
    rules = {
        "high_error_rate": {"condition": "error_rate > 0.05", "severity": "critical", "action": "notify_oncall"},
        "high_latency":    {"condition": "p99_latency > 30s", "severity": "warning",  "action": "notify_team"},
        "agent_timeout":   {"condition": "timeout_count > 10/min", "severity": "warning", "action": "investigate"},
    }

import os
from anthropic import Anthropic

client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

tools = [{
    "name": "get_weather",
    "description": "查询指定城市的天气",
    "input_schema": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}
}]

def execute_tool(tool_use):
    if tool_use.name == "get_weather":
        return f"{tool_use.input['city']} 今天晴，25°C"
    raise ValueError(f"未知工具: {tool_use.name}")

def run_agent(goal: str, max_iterations: int = 10):
    messages = [{"role": "user", "content": goal}]
    for i in range(max_iterations):
        response = client.messages.create(
            model="claude-opus-4-20250514", max_tokens=1024, tools=tools, messages=messages,
        )
        tool_blocks = [b for b in response.content if b.type == "tool_use"]
        if not tool_blocks:
            text = "".join(b.text for b in response.content if b.type == "text")
            print(f"[最终回答] {text}")
            return text
        messages.append({"role": "assistant", "content": response.content})
        results = []
        for tb in tool_blocks:
            result = execute_tool(tb)
            print(f"[工具调用] {tb.name}({tb.input}) → {result}")
            results.append({"type": "tool_result", "tool_use_id": tb.id, "content": result})
        messages.append({"role": "user", "content": results})
    print("达到最大迭代次数，Agent 终止")

if __name__ == "__main__":
    run_agent("帮我查一下北京和上海的天气，然后总结哪个更适合出行")

response = client.messages.create(
    model="claude-opus-4-20250514",
    messages=[{"role": "user", "content": "继续"}],
    tools=tools,
)