Here’s What Everyone Gets Wrong About Agentic AI

def compound_success_rate(per_step_accuracy: float, num_steps: int) -> float:
    """
    计算一个 n 步代理工作流程端到端成功的概率，
    给定每一步的准确率。基于可靠性工程中的 Lusser 定律。

    Args:
        per_step_accuracy: 每一步成功的概率（0.0 到 1.0）
        num_steps:          工作流程中的总步骤数

    Returns:
        总体成功概率，作为 0.0 到 1.0 之间的浮点数
    """
    return per_step_accuracy ** num_steps

# 在大多数生产代理实际运行的准确率范围内运行它
examples = [
    (0.95, 10, "95% 准确率，10 步工作流程"),
    (0.90, 10, "90% 准确率，10 步工作流程"),
    (0.85, 10, "85% 准确率，10 步工作流程"),
    (0.85,  3, "85% 准确率，3 步工作流程（狭窄范围）"),
]

for acc, steps, label in examples:
    rate = compound_success_rate(acc, steps)
    print(f"{label}: {rate * 100:.1f}% 总体成功率")

# 保存文件
python3 compound_reliability.py

95% 准确率，10步流程：总体成功率为59.9%
90% 准确率，10步流程：总体成功率为34.9%
85% 准确率，10步流程：总体成功率为19.7%
85% 准确率，3步流程（范围较窄）：总体成功率为61.4%

import json

# 一个最小化且类型明确的工具注册表。
# 关键设计原则：工具通过显式的模式进行定义，并标记为可逆或不可逆。代理本身不会决定这一点。

TOOLS = {
    "search_orders": {
        "description": "根据履约状态搜索客户订单。返回匹配的订单ID列表。",
        "irreversible": False,
        "inputSchema": {
            "type": "object",
            "properties": {
                "status": {
                    "type": "string",
                    "enum": ["pending", "shipped", "delivered", "cancelled"],
                    "description": "用于筛选订单的履约状态。"
                },
                "limit": {
                    "type": "integer",
                    "minimum": 1,
                    "maximum": 50,
                    "description": "返回结果的最大数量。"
                }
            },
            "required": ["status"]
        }
    },
    "cancel_order": {
        "description": "通过订单ID取消客户订单。此操作无法撤销。",
        "irreversible": True,   # 执行前硬性阻止；需要人工确认
        "inputSchema": {
            "type": "object",
            "properties": {
                "order_id": {
                    "type": "string",
                    "description": "要取消的订单的唯一标识符。"
                },
                "reason": {
                    "type": "string",
                    "description": "取消的原因。存储在审计日志中。"
                }
            },
            "required": ["order_id", "reason"]
        }
    },
    "send_confirmation_email": {
        "description": "向客户发送取消确认电子邮件。此操作无法撤销。",
        "irreversible": True,
        "inputSchema": {
            "type": "object",
            "properties": {
                "to": {"type": "string", "description": "客户电子邮件地址。"},
                "order_id": {"type": "string", "description": "要在电子邮件中包含的订单ID。"}
            },
            "required": ["to", "order_id"]
        }
    }
}

def validate_tool_input(tool_name: str, args: dict) -> bool:
    """
    验证参数是否匹配工具声明的输入模式。
    在执行前捕获错误的工具调用和格式错误的参数。
    如果验证失败，会引发带有清晰信息的ValueError。
    """
    if tool_name not in TOOLS:
        raise ValueError(
            f"未知工具: '{tool_name}'。可用工具: {list(TOOLS.keys())}"
        )

    schema = TOOLS[tool_name]["inputSchema"]
    required_fields = schema.get("required", [])
    defined_properties = schema.get("properties", {})

    # 检查所有必需字段是否都存在
    for field in required_fields:
        if field not in args:
            raise ValueError(
                f"工具 '{tool_name}' 缺少必需字段 '{field}'。"
            )

    # 验证枚举约束和类型
    for field, value in args.items():
        if field not in defined_properties:
            continue  # 允许存在额外字段，但不要引发错误；在生产环境中记录这些字段

        field_schema = defined_properties[field]

        if "enum" in field_schema and value not in field_schema["enum"]:
            raise ValueError(

def validate_tool_input(tool_name: str, args: dict) -> bool:
    """
    Validate that the input arguments for a tool are correct.
    Raises ValueError on invalid input.
    """
    tool = TOOLS[tool_name]
    for field, value in args.items():
        field_schema = tool["schema"][field]

        if "enum" in field_schema and value not in field_schema["enum"]:
            raise ValueError(
                f"Invalid value '{value}' for field '{field}' in tool '{tool_name}'. "
                f"Must be one of: {field_schema['enum']}"
            )

        if field_schema.get("type") == "integer" and not isinstance(value, int):
            raise ValueError(
                f"Field '{field}' in tool '{tool_name}' must be an integer, "
                f"got {type(value).__name__}."
            )

    return True

def execute_tool(tool_name: str, args: dict, human_confirmed: bool = False) -> dict:
    """
    Execute a tool with schema validation and human-in-the-loop gating
    for all irreversible actions.

    Returns a dict with:
        'result'           - the tool output string, or None if approval needed
        'requires_approval'- True if the call was halted for human review
        'message'          - explanation when approval is required
    """
    validate_tool_input(tool_name, args)

    tool = TOOLS[tool_name]

    # Gate on irreversibility -- this is the check that prevents database deletions,
    # unauthorized purchases, and emails sent to the wrong recipient.
    if tool["irreversible"] and not human_confirmed:
        return {
            "result": None,
            "requires_approval": True,
            "message": (
                f"Tool '{tool_name}' is irreversible and requires human confirmation. "
                f"Planned args: {json.dumps(args)}"
            )
        }

    # Safe to proceed -- replace this comment with your actual tool implementation
    return {
        "result": f"Tool '{tool_name}' executed successfully with args: {json.dumps(args)}",
        "requires_approval": False
    }

# --- Test runs ---

# 1. Valid reversible call -- executes immediately, no approval needed
response = execute_tool("search_orders", {"status": "shipped", "limit": 10})
print(f"Reversible tool:\n  {response['result']}\n")

# 2. Irreversible call without confirmation -- pauses and asks before doing anything
response = execute_tool("cancel_order", {"order_id": "ORD-12345", "reason": "Customer request"})
print(f"Irreversible without confirmation:")
print(f"  requires_approval = {response['requires_approval']}")
print(f"  message: {response['message']}\n")

# 3. Irreversible call with explicit confirmation -- proceeds normally
response = execute_tool(
    "cancel_order",
    {"order_id": "ORD-12345", "reason": "Customer request"},
    human_confirmed=True
)
print(f"Irreversible with confirmation:\n  {response['result']}\n")

# 4. Invalid enum value -- validation catches it before anything executes
try:
    execute_tool("search_orders", {"status": "lost"})
except ValueError as e:
    print(f"Invalid input caught:\n  {e}\n")

# 5. Missing required field -- caught before execution
try:
    execute_tool("cancel_order", {"order_id": "ORD-12345"})  # 'reason' is required
except ValueError as e:
    print(f"Missing field caught:\n  {e}")

python3 agent_tool_registry.py

可逆工具：
  工具 'search_orders' 已成功执行，参数为：{"status": "shipped", "limit": 10}

需要确认的不可逆操作：
  requires_approval = True
  消息：工具 'cancel_order' 是不可逆的，需要人工确认。计划参数：{"order_id": "ORD-12345", "reason": "Customer request"}

已确认的不可逆操作：
  工具 'cancel_order' 已成功执行，参数为：{"order_id": "ORD-12345", "reason": "Customer request"}

捕获到无效输入：
  工具 'search_orders' 的字段 'status' 值 'lost' 无效。必须是以下之一：['pending', 'shipped', 'delivered', 'cancelled']

捕获到缺失字段：
  工具 'cancel_order' 缺少必填字段 'reason'。

import json
import datetime

class AgentTracer:
    """
    记录代理在工作流运行过程中每次工具调用的完整追踪信息。
    在每一步中捕获输入、输出、延迟时间和一个置信度评分。

    这就是为什么在第3步就能发现失败，而不是等到第10步才发现问题，那时已经造成了损害。
    """

    def __init__(self, run_id: str):
        self.run_id = run_id
        self.steps = []

    def trace(
        self,
        step_index: int,
        tool_name: str,
        args: dict,
        result: str,
        latency_ms: float,
        confidence: float,
        low_confidence_threshold: float = 0.70,
    ) -> dict:
        """
        记录一次工具调用的完整上下文信息。

        Args:
            step_index:               工作流中的步骤编号（从1开始计数）
            tool_name:                被调用的工具名称
            args:                     传递给工具的参数
            result:                   工具的输出（在日志中被截断）
            latency_ms:               工具调用所用的时间（以毫秒为单位）
            confidence:               代理报告的置信度（0.0-1.0）
            low_confidence_threshold: 低于此置信度的步骤将被标记以供审查

        Returns:
            dict: 该步骤的完整追踪条目
        """
        entry = {
            "run_id": self.run_id,
            "step": step_index,
            "tool": tool_name,
            "args": args,
            # 截断长结果，使仪表板中的日志保持可读性
            "result_preview": result[:120] + "..." if len(result) > 120 else result,
            "latency_ms": round(latency_ms, 2),
            "confidence": round(confidence, 3),
            # 低于阈值的步骤将在运行摘要中被标记以供人工审查
            "low_confidence": confidence < low_confidence_threshold,
            "timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(),
        }
        self.steps.append(entry)
        return entry

    def summary(self) -> dict:
        """
        总结运行情况：总步骤数、总延迟时间和被标记的步骤。

        在你的运行后日志和警报流程中使用此功能。
        低置信度步骤是静默失败的早期预警信号。
        """
        total_latency = sum(s["latency_ms"] for s in self.steps)
        flagged = [s for s in self.steps if s["low_confidence"]]

        return {
            "run_id": self.run_id,
            "total_steps": len(self.steps),
            "total_latency_ms": round(total_latency, 2),
            "flagged_steps": len(flagged),
            "flagged_details": [
                {
                    "step": s["step"],
                    "tool": s["tool"],
                    "confidence": s["confidence"],
                }
                for s in flagged
            ],
        }

# 模拟一个包含完整追踪的5步客户支持代理工作流

tracer = AgentTracer(run_id="run-support-2026-001")

# 每个元组：(工具名称，参数，结果，延迟时间，置信度)
# 置信度低于0.70的步骤将在摘要中被自动标记。
simulated_steps = [
    (
        "search_orders",
        {"status": "pending"},
        "Found 3 pending orders: ORD-001, ORD-002, ORD-003",
        45.2,

simulated_steps = [
    (
        "create_order",
        {"customer_id": "CUST-123", "product_id": "WIDGET-A", "quantity": 2},
        "Order created: ORD-001",
        120.3,
        0.95,  # High confidence -- agent is certain about this step
    ),
    (
        "get_order_detail",
        {"order_id": "ORD-001"},
        "Order ORD-001: 2x Widget, $49.99, estimated delivery June 20",
        38.7,
        0.91,
    ),
    (
        "check_inventory",
        {"product_id": "WIDGET-A"},
        "WIDGET-A: 12 units in stock at Warehouse Lagos",
        210.5,
        0.61,  # LOW CONFIDENCE -- agent uncertain about warehouse location; flagged
    ),
    (
        "update_order",
        {"order_id": "ORD-001", "status": "confirmed"},
        "Order ORD-001 status updated to confirmed",
        55.1,
        0.88,
    ),
    (
        "send_confirmation_email",
        {"to": "customer@example.com", "order_id": "ORD-001"},
        "Email queued for delivery to customer@example.com",
        30.0,
        0.52,  # LOW CONFIDENCE -- agent uncertain about recipient; flagged before irreversible send
    ),
]

print("=== Step-by-step trace ===")
for i, (tool, args, result, latency, confidence) in enumerate(simulated_steps):
    entry = tracer.trace(i + 1, tool, args, result, latency, confidence)
    flag = " [LOW CONFIDENCE -- FLAGGED FOR REVIEW]" if entry["low_confidence"] else ""
    print(f"  Step {i + 1}: {tool}{flag}")

print("\n=== Run Summary ===")
print(json.dumps(tracer.summary(), indent=2))

python3 agent_tracer.py

=== 分步追踪 ===
  步骤 1: search_orders
  步骤 2: get_order_detail
  步骤 3: check_inventory [置信度较低 -- 需要审查]
  步骤 4: update_order
  步骤 5: send_confirmation_email [置信度较低 -- 需要审查]

=== 运行摘要 ===
{
  "run_id": "run-support-2026-001",
  "total_steps": 5,
  "total_latency_ms": 379.5,
  "flagged_steps": 2,
  "flagged_details": [
    {"step": 3, "tool": "check_inventory", "confidence": 0.61},
    {"step": 5, "tool": "send_confirmation_email", "confidence": 0.52}
  ]
}