flowchart TD
    A[试点启动<br/>38%企业已启动] --> B{第1关:可靠性关}
    B -->|通过| C{第2关:工具调用关}
    B -->|失败| X[❌ 项目死亡<br/>89%折损发生在此]
    C -->|通过| D{第3关:熔断防护关}
    C -->|失败| X
    D -->|通过| E{第4关:成本控制关}
    D -->|失败| X
    E -->|通过| F{第5关:可观测关}
    E -->|失败| X
    F -->|通过| G{第6关:安全合规关}
    F -->|失败| X
    G -->|通过| H{第7关:ROI验证关}
    G -->|失败| X
    H -->|通过| I[✅ 生产规模化<br/>11%存活]
    H -->|失败| X

    style X fill:#ff6b6b,color:#fff
    style I fill:#51cf66,color:#fff
    style B fill:#ffd43b
    style C fill:#ffd43b
    style D fill:#ffd43b
    style E fill:#ffd43b
    style F fill:#ffd43b
    style G fill:#ffd43b
    style H fill:#ffd43b

# Process Manager核心模式 - 来自生产实践
class AgentProcessManager:
    def __init__(self, agent, config):
        self.agent = agent
        self.max_retries = config.max_retries  # 通常3次
        self.confidence_threshold = 0.92
        self.checkpoint_interval = 3  # 每3步一次检查

    async def execute_with_guards(self, task):
        steps_completed = 0
        total_budget = task.max_steps

        while steps_completed < total_budget:
            # 执行一步
            result = await self.agent.step(task)
            steps_completed += 1

            # 检查点验证
            if steps_completed % self.checkpoint_interval == 0:
                validation = await self.validate_progress(task, result)
                if not validation.on_track:
                    # 回滚到最后一个检查点
                    await self.rollback_to_checkpoint(task)
                    # 注入纠正上下文
                    result = await self.agent.retry_with_correction(
                        task, validation.deviation_report
                    )

            # 置信度检查
            if result.confidence < self.confidence_threshold:
                await self.escalate_to_human(task, result)
                return HumanTakeoverResult(task, result)

        return result

# 工具调用合约验证引擎
class ToolCallContract:
    def __init__(self, tool_registry):
        self.registry = tool_registry
        self.contracts = self._load_contracts()

    async def validate_call(self, tool_name, params):
        contract = self.contracts.get(tool_name)
        if not contract:
            raise UnknownToolError(tool_name)

        errors = []
        # 1. 参数完整性校验
        for required_field in contract.required_params:
            if required_field not in params:
                errors.append(f"Missing required: {required_field}")

        # 2. 参数类型和格式校验
        for field, schema in contract.param_schemas.items():
            if field in params and not schema.validate(params[field]):
                errors.append(f"Invalid format: {field}")

        # 3. 语义校验 - 参数是否在合法范围内
        if contract.semantic_validator:
            semantic_errors = await contract.semantic_validator(params)
            errors.extend(semantic_errors)

        # 4. 副作用预检 - 检查调用是否会产生不可逆操作
        if contract.has_side_effects and not params.get("_confirmed"):
            return PreCheckResult(needs_confirmation=True, 
                                side_effects=contract.side_effects)

        return ValidationResult(valid=len(errors)==0, errors=errors)

flowchart LR
    L1[第1层:步骤熔断<br/>单步超时/重试上限] --> L2[第2层:任务熔断<br/>总步数/总时间上限]
    L2 --> L3[第3层:成本熔断<br/>Token/美元预算上限]
    L3 --> L4[第4层:系统熔断<br/>全局限流/紧急停止]

    L1 --> |触发| H1[重试+上下文压缩]
    L2 --> |触发| H2[状态保存+人类升级]
    L3 --> |触发| H3[强制终止+费用告警]
    L4 --> |触发| H4[全集群暂停+运维告警]

# 四层熔断器实现
class AgentCircuitBreaker:
    def __init__(self, config):
        # 第1层：步骤级
        self.step_timeout = config.step_timeout        # 默认30秒
        self.step_retry_limit = config.step_retries    # 默认3次

        # 第2层：任务级
        self.task_max_steps = config.task_max_steps    # 默认15步
        self.task_max_duration = config.task_duration  # 默认10分钟

        # 第3层：成本级
        self.token_budget = config.token_budget        # 默认100K tokens
        self.cost_budget = config.cost_budget          # 默认$2/任务

        # 第4层：系统级
        self.global_rpm_limit = config.rpm_limit       # 默认60 RPM
        self.emergency_kill_switch = False

    async def check_step(self, step_result):
        """第1层检查：单步是否超时或失败"""
        if step_result.duration > self.step_timeout:
            return CircuitOpen("step_timeout", 
                              action="retry_with_compressed_context")
        if step_result.retry_count >= self.step_retry_limit:
            return CircuitOpen("step_retry_exhausted",
                              action="skip_and_log")
        return CircuitClosed()

    async def check_task(self, task_state):
        """第2层检查：任务是否超出边界"""
        if task_state.steps >= self.task_max_steps:
            return CircuitOpen("task_steps_exceeded",
                              action="save_state_and_escalate")
        if task_state.duration > self.task_max_duration:
            return CircuitOpen("task_duration_exceeded",
                              action="save_state_and_escalate")
        return CircuitClosed()

    async def check_budget(self, usage):
        """第3层检查：成本是否超限"""
        if usage.total_tokens > self.token_budget:
            return CircuitOpen("token_budget_exceeded",
                              action="force_terminate")
        if usage.total_cost > self.cost_budget:
            return CircuitOpen("cost_budget_exceeded",
                              action="force_terminate_and_alert")
        return CircuitClosed()

# Agent全链路Trace
@trace(span_type="agent_task")
async def execute_agent_task(task):
    with span("planning") as plan_span:
        plan = await agent.plan(task)
        plan_span.set_attributes({
            "plan.steps": len(plan.steps),
            "plan.estimated_tokens": plan.token_estimate,
            "plan.confidence": plan.confidence
        })

    for i, step in enumerate(plan.steps):
        with span(f"step_{i}", span_type="agent_step") as step_span:
            # 工具调用Trace
            with span("tool_call", span_type="tool") as tool_span:
                tool_result = await execute_tool(step.tool, step.params)
                tool_span.set_attributes({
                    "tool.name": step.tool,
                    "tool.params_hash": hash(str(step.params)),
                    "tool.duration_ms": tool_result.duration,
                    "tool.success": tool_result.success
                })

            # LLM推理Trace
            with span("llm_inference") as llm_span:
                inference = await agent.reason(step, tool_result)
                llm_span.set_attributes({
                    "llm.tokens_in": inference.input_tokens,
                    "llm.tokens_out": inference.output_tokens,
                    "llm.model": inference.model_version
                })

            step_span.set_attributes({
                "step.result": inference.decision,
                "step.cumulative_cost": running_cost
            })

    # 业务指标
    set_business_metrics({
        "task.completed": True,
        "task.total_cost": total_cost,
        "task.total_duration": total_duration,
        "task.human_escalated": False
    })

flowchart TB
    subgraph 网络隔离层
        A[Agent进程] -->|白名单网络| B[允许的API端点]
        A -.-x|阻断| C[未授权网络]
    end

    subgraph 文件系统隔离层
        D[Agent工作目录] -->|读写| E[/tmp/agent_workspace/]
        D -.-x|阻断| F[系统文件/用户数据]
    end

    subgraph 权限隔离层
        G[最小权限集] -->|仅调用| H[白名单工具]
        G -.-x|阻断| I[危险操作<br/>rm/send_email/支付]
    end

flowchart LR
    A[100% 启动试点] -->|可靠性质疑| B[65% 通过第1关]
    B -->|工具调用失败| C[52% 通过第2关]
    C -->|缺少熔断| D[44% 通过第3关]
    D -->|成本失控| E[39% 通过第4关]
    E -->|可观测性不足| F[36% 通过第5关]
    F -->|安全合规| G[34% 通过第6关]
    G -->|ROI不明| H[11% 通过第7关<br/>实现规模化]

    style A fill:#4dabf7,color:#fff
    style H fill:#51cf66,color:#fff