什么是 ‘Dry-run Mode’？为你的 Agent 工具箱实现一个‘预览模式’，防止非预期的副作用产生

各位同仁，下午好！

今天，我们聚焦一个在自动化和智能代理领域至关重要的概念——“Dry-run Mode”，中文常译为“试运行模式”或“空跑模式”。随着人工智能和自动化技术日益成熟，智能代理（Agent）被赋予了越来越多的自主权，它们能够理解复杂指令，调用各种工具（Tools）与外部环境交互，执行任务。这种能力固然强大，但也带来了一个潜在的风险：非预期的副作用。一个看似无害的指令，在代理的推理或工具的执行过程中，可能会因为各种原因（如逻辑错误、配置失误、环境差异等）导致不可逆的破坏性后果。

想象一下，一个基础设施管理代理错误地删除了生产数据库；一个数据处理代理覆盖了关键数据；一个安全代理在没有充分验证的情况下修改了防火墙规则。这些都是我们力图避免的灾难性场景。

Dry-run Mode正是解决这一问题的核心机制。它允许我们预演代理将要执行的操作，模拟其与外部世界的交互，并展示这些操作可能产生的影响，而无需真正执行任何有副作用的行为。它就像一个沙盒，提供了一个安全的空间来验证代理的意图和工具的执行路径。

本次讲座，我将深入探讨Dry-run Mode的原理、设计哲学，并为我们的Agent工具箱提供一套详细的实现方案，旨在构建一个健壮、可信赖的预览机制。

一、 Dry-run Mode：概念与核心价值

1.1 什么是 Dry-run Mode？

Dry-run Mode，顾名思义，是一种模拟执行模式。在这种模式下，系统会按照正常流程进行计算、决策和调度，但所有涉及修改外部状态（如文件系统、数据库、外部API、网络配置等）的操作都会被拦截或替换为无副作用的模拟行为。其核心目标是：

• 预测行为： 准确地报告如果真的执行，会发生什么。
• 预防风险： 避免在生产环境中造成意外或破坏性的更改。
• 提供反馈： 向用户或开发者展示操作的详细计划和预期结果。

它并非简单地跳过执行，而是要尽可能地模拟执行过程中的每一个步骤，包括参数解析、条件判断、工具选择等，直到即将触发实际副作用的那一刻，才将其替换为模拟的输出或报告。

1.2 Dry-run Mode 为何至关重要？

在Agent领域，Dry-run Mode的重要性尤为突出，主要体现在以下几个方面：

• 安全性保障： 这是最直接也是最重要的价值。它充当了一道防火墙，防止代理在不确定性下执行危险操作。尤其是在Agent的推理能力仍有局限、可能产生“幻觉”或误解指令时，Dry-run提供了一个关键的验证环节。
• 可预测性增强： 代理的自主性意味着其行为路径可能不总是完全可预测。Dry-run Mode能够清晰地展现代理将采取的每一步行动，包括调用哪些工具、传递哪些参数，以及预期的结果，从而大大增强了系统的可预测性。
• 调试与开发效率： 开发者可以在不担心破坏环境的情况下，快速迭代和测试代理的新功能或新工具。通过Dry-run的输出，可以更容易地发现代理逻辑或工具实现中的错误。
• 信任度建立： 对于用户而言，能够预览一个自动化系统将要做的所有事情，是建立信任的关键。用户可以审查代理的计划，并在确认无误后才授权其执行。
• 成本控制： 某些外部API调用可能会产生费用（如云资源创建、短信发送等）。Dry-run Mode可以模拟这些付费操作，避免不必要的成本支出。
• 合规性与审计： 在某些行业，对系统变更进行预审是合规性要求。Dry-run日志可以作为变更审查的重要依据。

1.3 Dry-run Mode 的应用场景

Dry-run Mode并非Agent领域的独有概念，它广泛应用于各种自动化和管理系统中：

• 命令行工具： 许多Linux命令（如rm -i，apt-get --dry-run）都提供了某种形式的预览功能。terraform plan是基础设施即代码（IaC）领域的经典Dry-run示例。
• 数据库管理： 数据库迁移工具通常提供Dry-run选项，以预览SQL脚本将要执行的更改。
• CI/CD 流水线： 在部署或发布前，可以运行Dry-run来验证部署脚本的正确性。
• 配置管理： Ansible、Puppet等工具的--check或--noop模式。
• 网络设备管理： 更改路由器或防火墙配置前，预览配置将如何影响网络。
• 智能代理： 这正是我们今天关注的焦点，代理在执行文件操作、API调用、数据库交互、系统命令等之前，提供详细的预览。

二、 Agent 工具箱的核心问题：非预期副作用

我们的智能代理通过调用一系列“工具”来与外部世界互动。这些工具封装了特定的功能，例如：

• 文件系统工具： FileReadTool、FileWriteTool、FileDeleteTool
• API 交互工具： APICallTool、DatabaseQueryTool、CloudResourceManageTool
• 系统命令工具： ShellCommandTool
• 沟通工具： EmailSendTool、MessagePostTool

这些工具的强大之处在于它们能够实际地改变外部环境。然而，这种能力也正是风险的来源。

2.1 代理行为的复杂性与不确定性

• 推理链的脆弱性： 代理的决策过程可能涉及复杂的推理链。链条中任何一环的错误（例如对用户指令的误解、对工具描述的误读、参数生成错误），都可能导致最终调用的工具及其参数偏离预期。
• 工具的副作用： 大多数工具被设计用来产生实际的副作用。例如，FileDeleteTool会永久删除文件，APICallTool可能创建资源、发送通知或触发复杂的工作流。这些副作用一旦发生，往往难以撤销。
• 环境差异： 代理在开发、测试和生产环境中的行为可能因配置、数据、权限等差异而有所不同。在测试环境中看似无害的操作，在生产环境中可能造成严重后果。
• 并发与竞态： 多个代理或并发任务可能在同一资源上操作，Dry-run可以帮助发现潜在的竞态条件，尽管它不能完全模拟并发执行的复杂性。

2.2 典型的非预期副作用示例

为了更好地理解Dry-run Mode的价值，我们来看几个具体的非预期副作用场景：

场景	代理意图	潜在副作用	Dry-run 价值
文件管理	清理旧日志文件	错误地删除了生产数据文件（例如，正则表达式匹配错误）	会报告“将删除 /var/www/html/prod_data.csv”，而不是“将删除 /var/log/*.log”
数据库操作	更新用户配置	错误地更新了所有用户的配置，或删除了关键表	会报告“将执行 UPDATE users SET config='...' WHERE id='all'”
API 调用	创建一个测试环境的云资源	错误地在生产环境中创建了昂贵的云资源，或发送了大量通知	会报告“将调用 POST /api/prod/instances” 而不是 /api/dev/instances”
系统命令	重启某个服务	错误地重启了关键的共享服务，导致系统中断	会报告“将执行 sudo systemctl restart critical_service”
沟通工具	向特定用户发送通知	错误地向所有用户群发了内部测试通知	会报告“将向 [email protected] 发送邮件”

在这些情况下，Dry-run Mode能够提供一个预览，让开发者或操作人员有机会在实际破坏发生之前介入并纠正问题。

三、 设计 Dry-run Mode 的核心原则

要构建一个有效且可靠的Dry-run Mode，我们需要遵循一些关键的设计原则。

1. 隔离性 (Isolation)： 这是Dry-run模式最根本的原则。任何在Dry-run模式下执行的操作，都绝不能对真实世界产生任何可观测的副作用。所有的文件写入、数据库更新、API调用等都必须被拦截或重定向到模拟环境（如内存、虚拟文件系统）。
2. 透明性 (Transparency)： 用户必须清楚地知道当前Agent是在Dry-run模式下运行，还是在真实执行模式下运行。输出日志、命令行提示、UI元素等都应明确指示当前模式。
3. 完整性 (Completeness)： Dry-run应该尽可能地模拟完整的执行路径，包括代理的决策过程、工具的选择、参数的生成、内部逻辑的流转。它不应仅仅是简单地跳过最终的副作用操作，而是要模拟到触发副作用前的最后一刻。
4. 保真度 (Fidelity)： Dry-run的输出应该尽可能准确地反映真实执行时会发生什么。这意味着模拟结果应该包含与真实结果相近的结构和信息，例如，如果一个工具在真实执行时会返回一个资源ID，那么Dry-run也应该返回一个模拟的ID。
5. 可观察性 (Observability)： Dry-run模式下产生的所有模拟操作都应该被详细地记录下来，包括调用的工具名称、传递的参数、模拟的结果以及任何潜在的警告或错误。这些日志是审查和调试的关键。
6. 用户控制 (User Control)： 必须提供明确且易于使用的机制来开启或关闭Dry-run模式，并可能允许调整Dry-run的粒度（例如，某些操作可以真实执行，而另一些则必须模拟）。
7. 一致性 (Consistency)： 在Dry-run模式下，代理的内部状态（如果工具不修改外部状态，但修改了代理的内部状态）应该与真实执行模式下保持一致，以确保后续决策的正确性。

四、 为 Agent 工具箱实现 Dry-run Mode

现在，我们将深入探讨如何在Agent工具箱中实现Dry-run Mode。我们将构建一个简化的Agent框架，并在此基础上演示几种Dry-run策略。

4.1 核心组件定义

首先，我们定义Agent工具箱的基础组件：ToolOutput、BaseTool、Toolbox、ToolExecutor和SimpleAgent。

import json
import os
import time
from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional, Tuple
from datetime import datetime

# --- 1. 工具输出结构 ---
class ToolOutput:
    """
    Represents the output of a tool execution, whether real or simulated.
    """
    def __init__(self, success: bool, message: str, data: Optional[Dict[str, Any]] = None, is_dry_run: bool = False):
        self.success = success
        self.message = message
        self.data = data if data is not None else {}
        self.is_dry_run = is_dry_run

    def __str__(self):
        status = "SUCCESS" if self.success else "FAILURE"
        mode = "[DRY-RUN]" if self.is_dry_run else "[REAL]"
        data_str = json.dumps(self.data, indent=2, ensure_ascii=False) if self.data else "N/A"
        return f"{mode} Status: {status}n{mode} Message: {self.message}n{mode} Data: {data_str}"

    def __repr__(self):
        return f"ToolOutput(success={self.success}, message='{self.message}', is_dry_run={self.is_dry_run}, data={self.data})"

# --- 2. 基础工具接口 ---
class BaseTool(ABC):
    """
    Abstract base class for all tools in the agent's toolbox.
    Defines the interface for real execution and provides a default dry-run implementation.
    """
    def __init__(self, name: str, description: str):
        self.name = name
        self.description = description

    @abstractmethod
    def execute(self, **kwargs) -> ToolOutput:
        """
        Executes the tool's primary action, which may have real-world side effects.
        """
        pass

    def dry_run_execute(self, **kwargs) -> ToolOutput:
        """
        Provides a simulated execution result for dry-run mode.
        This default implementation provides a generic description of the intended action.
        Tools *should* override this for more specific and informative simulation if possible.
        """
        action_desc = f"Tool '{self.name}' would be executed with arguments: {json.dumps(kwargs, indent=2, ensure_ascii=False)}"
        print(f"  [DRY-RUN Default] Simulating tool '{self.name}'...")
        return ToolOutput(
            success=True,
            message=f"DRY-RUN: This tool would perform an action described as: '{self.description}'. Simulated action: {action_desc}",
            data={"simulated_action": action_desc, "tool_name": self.name, "args": kwargs},
            is_dry_run=True
        )

# --- 3. 工具箱 ---
class Toolbox:
    """
    Manages a collection of available tools for the agent.
    """
    def __init__(self, tools: List[BaseTool]):
        self._tools = {tool.name: tool for tool in tools}

    def get_tool(self, name: str) -> Optional[BaseTool]:
        """Retrieves a tool by its name."""
        return self._tools.get(name)

    def list_tools(self) -> List[Dict[str, str]]:
        """Lists all available tools with their names and descriptions."""
        return [{"name": tool.name, "description": tool.description} for tool in self._tools.values()]

# --- 4. 工具执行器 (核心 Dry-run 逻辑注入点) ---
class ToolExecutor:
    """
    Handles the execution of tools. This is where the dry-run logic is primarily managed.
    """
    def __init__(self, toolbox: Toolbox, dry_run: bool = False):
        self.toolbox = toolbox
        self.dry_run = dry_run
        self.dry_run_log: List[ToolOutput] = [] # Stores outputs only when in dry-run mode

    def execute_tool(self, tool_name: str, **kwargs) -> ToolOutput:
        """
        Executes a specified tool. If in dry-run mode, it calls the tool's dry_run_execute method.
        Otherwise, it calls the real execute method.
        """
        tool = self.toolbox.get_tool(tool_name)
        if not tool:
            return ToolOutput(success=False, message=f"Error: Tool '{tool_name}' not found.")

        print(f"n{'[DRY-RUN]' if self.dry_run else '[REAL-EXECUTION]'} Agent attempting to call tool: {tool_name}")
        print(f"  Arguments: {json.dumps(kwargs, indent=2, ensure_ascii=False)}")

        if self.dry_run:
            output = tool.dry_run_execute(**kwargs)
            self.dry_run_log.append(output) # Log dry-run actions
            return output
        else:
            try:
                output = tool.execute(**kwargs)
                return output
            except Exception as e:
                return ToolOutput(success=False, message=f"Error during real execution of '{tool_name}': {e}")

    def get_dry_run_log(self) -> List[ToolOutput]:
        """Returns the accumulated log of dry-run actions."""
        return self.dry_run_log

    def clear_dry_run_log(self):
        """Clears the dry-run log."""
        self.dry_run_log = []

# --- 5. 简单的Agent示例 ---
class SimpleAgent:
    """
    A very simple agent that decides which tools to call based on a task description.
    For demonstration purposes, its "reasoning" is hardcoded.
    """
    def __init__(self, executor: ToolExecutor):
        self.executor = executor

    def plan_and_execute(self, task: str) -> List[ToolOutput]:
        print(f"n{'='*50}nAgent received task: '{task}'n{'='*50}")
        results: List[ToolOutput] = []

        # Simulate agent's reasoning based on task description
        if "create a report" in task.lower():
            report_content = f"Report for task: '{task}'.nGenerated on: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}n" 
                             f"This report covers key activities and findings related to the task."
            # Call FileWriteTool
            results.append(self.executor.execute_tool(
                "FileWriteTool",
                file_path="task_report.txt",
                content=report_content
            ))
            # Call APICallTool
            api_data = {
                "title": f"Agent Task Report: {task}",
                "author": "SimpleAgent",
                "timestamp": datetime.now().isoformat(),
                "content_summary": report_content[:100] + "..." if len(report_content) > 100 else report_content
            }
            results.append(self.executor.execute_tool(
                "APICallTool",
                endpoint="/api/v1/reports",
                method="POST",
                headers={"Content-Type": "application/json"},
                body=api_data
            ))
        elif "delete old log files" in task.lower():
            # Call FileDeleteTool for multiple files
            results.append(self.executor.execute_tool(
                "FileDeleteTool",
                file_path="old_log_2023.txt"
            ))
            results.append(self.executor.execute_tool(
                "FileDeleteTool",
                file_path="temp_data.csv"
            ))
            # Add a potentially dangerous delete to show Dry-run's value
            results.append(self.executor.execute_tool(
                "FileDeleteTool",
                file_path="/var/www/html/critical_prod_config.ini" # DANGEROUS!
            ))
        elif "update system configuration" in task.lower():
            results.append(self.executor.execute_tool(
                "SystemConfigUpdateTool",
                config_key="security_level",
                config_value="high"
            ))
            results.append(self.executor.execute_tool(
                "SystemConfigUpdateTool",
                config_key="max_connections",
                config_value=500
            ))
        else:
            results.append(ToolOutput(success=False, message=f"Agent cannot handle task: '{task}'"))

        return results

4.2 Dry-run 策略：混合方法

我们采用一种混合策略，它结合了“工具层抽象”和“代理执行器拦截”的优点：

• 默认 Dry-run 行为： BaseTool 提供一个通用的 dry_run_execute 方法，用于描述工具将要被调用的事实。
• 工具特定 Dry-run： 具体的工具可以（也应该）重写 dry_run_execute 方法，以提供更精确、更详细的模拟。
• 执行器拦截： ToolExecutor 负责根据 dry_run 标志，决定调用工具的 execute 还是 dry_run_execute 方法。

这种方法的好处是：对于简单的工具，开发者无需关心Dry-run实现；对于有复杂副作用的工具，可以精确地模拟其行为；同时，核心的Dry-run切换逻辑集中在ToolExecutor中，易于管理。

4.3 实现具体的工具

现在，我们来实现几个具有实际副作用的工具，并为它们提供（或不提供）自定义的dry_run_execute实现。

# --- 具体工具实现 ---

class FileWriteTool(BaseTool):
    def __init__(self):
        super().__init__(
            name="FileWriteTool",
            description="Writes content to a specified file path. Creates the file if it doesn't exist, overwrites if it does."
        )

    def execute(self, file_path: str, content: str) -> ToolOutput:
        """Real execution: writes content to the file system."""
        try:
            with open(file_path, "w", encoding="utf-8") as f:
                f.write(content)
            print(f"  [REAL] Successfully wrote {len(content)} bytes to '{file_path}'.")
            return ToolOutput(success=True, message=f"Content successfully written to '{file_path}'.", data={"file_path": file_path, "bytes_written": len(content)})
        except IOError as e:
            return ToolOutput(success=False, message=f"Failed to write to file '{file_path}': {e}")

    def dry_run_execute(self, file_path: str, content: str) -> ToolOutput:
        """Dry-run: describes what would be written without actually writing."""
        simulated_content_preview = content[:100] + "..." if len(content) > 100 else content
        message = (f"DRY-RUN: Would write {len(content)} bytes of content "
                   f"to file '{file_path}'. Preview of content: '{simulated_content_preview}'")
        print(f"  [DRY-RUN Custom] Simulating FileWriteTool: {message}")
        return ToolOutput(
            success=True,
            message=message,
            data={"file_path": file_path, "would_write_bytes": len(content), "content_preview": simulated_content_preview},
            is_dry_run=True
        )

class FileDeleteTool(BaseTool):
    def __init__(self):
        super().__init__(
            name="FileDeleteTool",
            description="Deletes a specified file path."
        )

    def execute(self, file_path: str) -> ToolOutput:
        """Real execution: deletes a file from the file system."""
        try:
            if os.path.exists(file_path):
                os.remove(file_path)
                print(f"  [REAL] Successfully deleted file '{file_path}'.")
                return ToolOutput(success=True, message=f"File '{file_path}' successfully deleted.", data={"file_path": file_path})
            else:
                print(f"  [REAL] File '{file_path}' does not exist, no action needed.")
                return ToolOutput(success=True, message=f"File '{file_path}' does not exist, no deletion performed.", data={"file_path": file_path, "status": "not_found"})
        except OSError as e:
            return ToolOutput(success=False, message=f"Failed to delete file '{file_path}': {e}")

    def dry_run_execute(self, file_path: str) -> ToolOutput:
        """Dry-run: describes that the file would be deleted."""
        # We can add more intelligence here, e.g., check if file exists in current *real* state
        # but still not delete it. For simplicity, we just report the intent.
        message = f"DRY-RUN: Would attempt to delete file '{file_path}'."
        exists_in_real = os.path.exists(file_path)
        if exists_in_real:
            message += " (File currently exists in real system)."
        else:
            message += " (File does not currently exist in real system, would result in no-op or error)."

        print(f"  [DRY-RUN Custom] Simulating FileDeleteTool: {message}")
        return ToolOutput(
            success=True,
            message=message,
            data={"file_path": file_path, "would_delete": True, "exists_in_real_system": exists_in_real},
            is_dry_run=True
        )

class APICallTool(BaseTool):
    def __init__(self):
        super().__init__(
            name="APICallTool",
            description="Makes an HTTP API call to a specified endpoint with given method, headers, and body."
        )

    def execute(self, endpoint: str, method: str = "GET", headers: Optional[Dict[str, str]] = None, body: Optional[Dict[str, Any]] = None) -> ToolOutput:
        """Real execution: would make an actual HTTP request. For demo, we simulate a network delay."""
        print(f"  [REAL] Making a real API call to {endpoint} ({method})... (Simulating network delay)")
        time.sleep(0.5) # Simulate network delay
        # In a real scenario, you'd use 'requests' library here.
        # For this demo, we'll just return a mock success.
        mock_response_data = {"status": "success", "resource_id": f"res_{int(time.time())}", "message": "Resource created/updated."}
        print(f"  [REAL] API call to {endpoint} completed with mock success.")
        return ToolOutput(
            success=True,
            message=f"API call to '{endpoint}' ({method}) successfully made.",
            data={"endpoint": endpoint, "method": method, "response": mock_response_data},
        )

    def dry_run_execute(self, endpoint: str, method: str = "GET", headers: Optional[Dict[str, str]] = None, body: Optional[Dict[str, Any]] = None) -> ToolOutput:
        """Dry-run: describes the API call without making it."""
        message = (f"DRY-RUN: Would make an API call to '{endpoint}' using method '{method}'. "
                   f"Headers: {headers}. Body: {json.dumps(body, ensure_ascii=False) if body else 'N/A'}.")
        mock_response_data = {"status": "simulated_success", "simulated_resource_id": "sim_res_12345", "message": "Simulated resource creation/update."}
        print(f"  [DRY-RUN Custom] Simulating APICallTool: {message}")
        return ToolOutput(
            success=True,
            message=message,
            data={"endpoint": endpoint, "method": method, "simulated_response": mock_response_data},
            is_dry_run=True
        )

class SystemConfigUpdateTool(BaseTool):
    def __init__(self):
        super().__init__(
            name="SystemConfigUpdateTool",
            description="Updates a system configuration key with a new value. This could be a critical operation."
        )

    def execute(self, config_key: str, config_value: Any) -> ToolOutput:
        """Real execution: updates the actual system configuration (mocked here)."""
        print(f"  [REAL] Updating system config: {config_key} = {config_value}...")
        # In a real system, this would interact with OS/DB config files or a config service
        time.sleep(0.2)
        print(f"  [REAL] System config for '{config_key}' updated to '{config_value}'.")
        return ToolOutput(
            success=True,
            message=f"System configuration '{config_key}' updated to '{config_value}'.",
            data={"config_key": config_key, "new_value": config_value}
        )

    def dry_run_execute(self, config_key: str, config_value: Any) -> ToolOutput:
        """Dry-run: describes the configuration change."""
        message = (f"DRY-RUN: Would update system configuration key '{config_key}' "
                   f"to new value '{config_value}'.")
        print(f"  [DRY-RUN Custom] Simulating SystemConfigUpdateTool: {message}")
        return ToolOutput(
            success=True,
            message=message,
            data={"config_key": config_key, "would_change_to": config_value},
            is_dry_run=True
        )

4.4 运行演示

我们将通过两个场景来演示Dry-run Mode：

1. Dry-run 模式： 代理将模拟执行任务，不会产生任何实际副作用。
2. 真实执行模式： 代理将尝试执行实际操作，可能会创建、修改或删除文件，并模拟API调用。

# --- 演示主程序 ---

def setup_environment():
    """Ensures a clean slate for file operations."""
    print("n--- Setting up environment ---")
    if os.path.exists("task_report.txt"):
        os.remove("task_report.txt")
        print("  Cleaned up 'task_report.txt'")
    if os.path.exists("old_log_2023.txt"):
        os.remove("old_log_2023.txt")
        print("  Cleaned up 'old_log_2023.txt'")
    if os.path.exists("temp_data.csv"):
        os.remove("temp_data.csv")
        print("  Cleaned up 'temp_data.csv'")
    # Create some files for deletion demonstration
    with open("old_log_2023.txt", "w") as f:
        f.write("This is an old log file.")
    with open("temp_data.csv", "w") as f:
        f.write("col1,col2n1,an2,b")
    # This dangerous file will exist for Dry-run to report, but not be created in real run to prevent actual damage
    # if os.path.exists("/var/www/html/critical_prod_config.ini"):
    #     os.remove("/var/www/html/critical_prod_config.ini")
    # with open("/var/www/html/critical_prod_config.ini", "w") as f:
    #     f.write("[prod]ndb_host=prod_db")
    print("  Environment ready. 'old_log_2023.txt' and 'temp_data.csv' created for deletion tests.")

def run_demonstration():
    # Setup the toolbox with all our tools
    toolbox = Toolbox(tools=[
        FileWriteTool(),
        FileDeleteTool(),
        APICallTool(),
        SystemConfigUpdateTool()
    ])

    print("n" + "#" * 80)
    print("### Dry-run Mode Demonstration ###")
    print("#" * 80)

    # --- Scenario 1: Dry-run Mode ---
    print("n" + "="*30 + " RUNNING IN DRY-RUN MODE " + "="*30)
    setup_environment() # Ensure clean state for each run
    dry_run_executor = ToolExecutor(toolbox=toolbox, dry_run=True)
    dry_run_agent = SimpleAgent(executor=dry_run_executor)

    dry_run_tasks = [
        "Please create a report summarizing today's activities.",
        "I need you to delete old log files and temporary data.",
        "Update system configuration to enhance security."
    ]

    for task in dry_run_tasks:
        dry_run_results = dry_run_agent.plan_and_execute(task)
        print("n--- Dry-run Results Summary ---")
        for output in dry_run_results:
            print(output)
        print("-" * 30)

    print("n--- Full Dry-run Log ---")
    for i, log_entry in enumerate(dry_run_executor.get_dry_run_log()):
        print(f"Log Entry {i+1}:n{log_entry}n")
    print("="*80)
    print("Dry-run complete. No actual changes were made to the file system or external systems.")
    print("You can inspect the dry_run_log to see what *would have* happened.")
    print("="*80)

    # --- Scenario 2: Real Execution Mode ---
    print("nn" + "#" * 80)
    print("### Real Execution Mode Demonstration ###")
    print("#" * 80)

    # IMPORTANT: For real execution, be cautious. We'll clean up afterwards.
    print("n" + "="*30 + " RUNNING IN REAL EXECUTION MODE " + "="*30)
    print("!!! CAUTION: This will perform actual file operations and simulate API calls. !!!")
    input("Press Enter to proceed with REAL EXECUTION (or Ctrl+C to abort)...")

    setup_environment() # Re-setup environment for real run
    real_executor = ToolExecutor(toolbox=toolbox, dry_run=False)
    real_agent = SimpleAgent(executor=real_executor)

    real_tasks = [
        "Please create a report summarizing today's activities.",
        "I need you to delete old log files and temporary data."
        # Not including "update system configuration" in real run to avoid potential issues
    ]

    for task in real_tasks:
        real_results = real_agent.plan_and_execute(task)
        print("n--- Real Execution Results Summary ---")
        for output in real_results:
            print(output)
        print("-" * 30)

    print("n--- Verifying Real Execution ---")
    if os.path.exists("task_report.txt"):
        print(f"  'task_report.txt' exists. Content preview:")
        with open("task_report.txt", "r", encoding="utf-8") as f:
            print(f.read(200) + "...")
    else:
        print("  'task_report.txt' does NOT exist (unexpected).")

    if not os.path.exists("old_log_2023.txt"):
        print("  'old_log_2023.txt' was successfully deleted.")
    else:
        print("  'old_log_2023.txt' still exists (unexpected).")

    if not os.path.exists("temp_data.csv"):
        print("  'temp_data.csv' was successfully deleted.")
    else:
        print("  'temp_data.csv' still exists (unexpected).")

    # The dangerous file should not have been deleted in real execution because it was not created.
    # If it were created for test, Dry-run would still have warned.
    print("  '/var/www/html/critical_prod_config.ini' was never created/touched in this demo.")

    print("n" + "="*80)
    print("Real execution complete. Actual changes were made. Please inspect your environment.")
    print("="*80)

if __name__ == "__main__":
    run_demonstration()

4.5 演示输出分析

在Dry-run模式下，你会看到大量的[DRY-RUN]前缀输出，并且FileWriteTool和FileDeleteTool的dry_run_execute方法会提供详细的“将要执行”的描述，包括文件路径、内容预览等。最关键的是，即使代理“尝试”删除/var/www/html/critical_prod_config.ini，Dry-run模式也会明确指出这个意图，而不会真正触及该文件。

# (部分 Dry-run 模式输出示例)

==================================================
Agent received task: 'Please create a report summarizing today's activities.'
==================================================

[DRY-RUN] Agent attempting to call tool: FileWriteTool
  Arguments: {
  "file_path": "task_report.txt",
  "content": "Report for task: 'Please create a report summarizing today's activities.'.nGenerated on: 2023-10-27 15:30:00nThis report covers key activities and findings related to the task."
}
  [DRY-RUN Custom] Simulating FileWriteTool: DRY-RUN: Would write 200 bytes of content to file 'task_report.txt'. Preview of content: 'Report for task: 'Please create a report summarizing today's activities.'.
Generated on: 2023-10-27 15:30:00
This report '
[DRY-RUN] Status: SUCCESS
[DRY-RUN] Message: DRY-RUN: Would write 200 bytes of content to file 'task_report.txt'. Preview of content: 'Report for task: 'Please create a report summarizing today's activities.'.
Generated on: 2023-10-27 15:30:00
This report '
[DRY-RUN] Data: {
  "file_path": "task_report.txt",
  "would_write_bytes": 200,
  "content_preview": "Report for task: 'Please create a report summarizing today's activities.'.nGenerated on: 2023-10-27 15:30:00nThis report "
}

[DRY-RUN] Agent attempting to call tool: APICallTool
  Arguments: {
  "endpoint": "/api/v1/reports",
  "method": "POST",
  "headers": {
    "Content-Type": "application/json"
  },
  "body": {
    "title": "Agent Task Report: Please create a report summarizing today's activities.",
    "author": "SimpleAgent",
    "timestamp": "2023-10-27T15:30:00.123456",
    "content_summary": "Report for task: 'Please create a report summarizing today's activities.'.nGenerated on: 2023-10-27 15:30:00nThis report covers key activities and findings related to the task."
  }
}
  [DRY-RUN Custom] Simulating APICallTool: DRY-RUN: Would make an API call to '/api/v1/reports' using method 'POST'. Headers: {'Content-Type': 'application/json'}. Body: {"title": "Agent Task Report: Please create a report summarizing today's activities.", "author": "SimpleAgent", "timestamp": "2023-10-27T15:30:00.123456", "content_summary": "Report for task: 'Please create a report summarizing today's activities.'.nGenerated on: 2023-10-27 15:30:00nThis report covers key activities and findings related to the task."}.
[DRY-RUN] Status: SUCCESS
[DRY-RUN] Message: DRY-RUN: Would make an API call to '/api/v1/reports' using method 'POST'. Headers: {'Content-Type': 'application/json'}. Body: {"title": "Agent Task Report: Please create a report summarizing today's activities.", "author": "SimpleAgent", "timestamp": "2023-10-27T15:30:00.123456", "content_summary": "Report for task: 'Please create a report summarizing today's activities.'.nGenerated on: 2023-10-27 15:30:00nThis report covers key activities and findings related to the task."}.
[DRY-RUN] Data: {
  "endpoint": "/api/v1/reports",
  "method": "POST",
  "simulated_response": {
    "status": "simulated_success",
    "simulated_resource_id": "sim_res_12345",
    "message": "Simulated resource creation/update."
  }
}

... (其他 Dry-run 任务输出)

[DRY-RUN] Agent attempting to call tool: FileDeleteTool
  Arguments: {
  "file_path": "/var/www/html/critical_prod_config.ini"
}
  [DRY-RUN Custom] Simulating FileDeleteTool: DRY-RUN: Would attempt to delete file '/var/www/html/critical_prod_config.ini'. (File does not currently exist in real system, would result in no-op or error).
[DRY-RUN] Status: SUCCESS
[DRY-RUN] Message: DRY-RUN: Would attempt to delete file '/var/www/html/critical_prod_config.ini'. (File does not currently exist in real system, would result in no-op or error).
[DRY-RUN] Data: {
  "file_path": "/var/www/html/critical_prod_config.ini",
  "would_delete": true,
  "exists_in_real_system": false
}

在真实执行模式下，你将看到[REAL-EXECUTION]前缀的输出，并且文件系统上确实会创建task_report.txt，同时old_log_2023.txt和temp_data.csv会被删除。

通过对比两种模式的输出，我们可以清晰地看到Dry-run Mode是如何在不产生实际副作用的情况下，提供关于Agent行为的全面洞察。

五、 高级考量与最佳实践

上述实现提供了一个基础且实用的Dry-run框架。但在更复杂的Agent系统中，我们还需要考虑以下高级问题：

5.1 状态模拟的挑战

Dry-run模式的“保真度”很大程度上取决于其对外部状态变化的模拟能力。

• 虚拟文件系统： 对于文件操作，可以使用内存中的文件系统库（如Python的pyfakefs）来捕获文件读写，而不触及真实磁盘。
• 数据库事务与回滚： 模拟数据库更新更具挑战性。理想情况下，Dry-run应该在一个可回滚的事务中执行所有数据库操作，并在Dry-run结束后回滚。或者使用一个内存中的SQLite数据库作为模拟目标。
• API Mocking： 对于外部API调用，可以使用requests-mock、responses等库来拦截HTTP请求，并返回预设的模拟响应，而不是真正发起网络请求。
• 环境快照： 在Dry-run开始前，对关键系统状态进行快照，并在Dry-run期间模拟基于该快照的变更。

5.2 局部 Dry-run 与粒度控制

并非所有工具在Dry-run模式下都必须完全模拟。有时，我们希望某些只读操作（如读取配置、查询状态）能够真实执行，因为它们的输出可能影响后续的代理决策，且本身没有副作用。

• 工具级别标志： 在BaseTool或其子类中引入一个always_real_in_dry_run或read_only_in_dry_run的标志。
• 上下文感知执行： ToolExecutor可以传递一个更丰富的ExecutionContext对象给工具，该对象包含is_dry_run以及更细粒度的控制标志。工具内部可以根据这些标志决定是执行真实读操作，还是完全模拟。

# 示例：上下文感知执行的思路

class ExecutionContext:
    def __init__(self, dry_run: bool, allow_real_reads_in_dry_run: bool = True):
        self.dry_run = dry_run
        self.allow_real_reads_in_dry_run = allow_real_reads_in_dry_run

class ContextAwareTool(BaseTool):
    # ...
    def execute_with_context(self, context: ExecutionContext, **kwargs) -> ToolOutput:
        if context.dry_run:
            if self.is_read_only_tool and context.allow_real_reads_in_dry_run:
                # Perform real read operation
                return self._perform_real_read(**kwargs)
            else:
                # Perform dry-run simulation
                return self.dry_run_execute(**kwargs)
        else:
            # Always real execution
            return self.execute(**kwargs)

5.3 报告与可视化

Dry-run的输出不仅仅是日志。一个好的Dry-run系统应该提供结构化、易于理解的报告：

• JSON/YAML 格式输出： 便于机器解析和集成到其他系统。
• 人类可读的总结： 清晰地列出所有“将要发生”的关键操作。
• 差异报告 (Diff)： 对于文件或配置的更改，显示Dry-run生成的更改与当前系统状态之间的差异，就像git diff或terraform plan一样。
• 依赖图： 如果Agent的工具调用是链式的，可以可视化工具调用的顺序和依赖关系。

5.4 与测试框架集成

Dry-run模式本质上是一种强大的集成测试手段。它可以无缝地集成到单元测试、集成测试和端到端测试流程中：

• 自动化验证： 编写测试用例，断言Dry-run日志中包含了预期的操作描述和参数。
• 回归测试： 确保Agent或工具的修改不会意外改变Dry-run的输出，从而保证其行为的一致性。

5.5 性能考量

虽然Dry-run避免了实际的副作用，但如果模拟过程过于复杂或资源密集，也可能影响性能。需要平衡保真度和执行效率。对于性能敏感的场景，可能需要简化Dry-run的模拟逻辑。

5.6 安全性与审计

Dry-run日志本身也可能包含敏感信息（例如，将要发送的API密钥、文件名、数据内容）。在Dry-run日志中，应注意对敏感信息进行脱敏或加密。此外，Dry-run的开启和关闭也应有适当的权限控制。

六、 Dry-run Mode 的实际应用场景示例

除了我们前面提到的通用示例，在Agent和自动化领域，Dry-run Mode的实际应用非常广泛：

• DevOps Agent： 在部署新服务或更新基础设施之前，Agent可以先以Dry-run模式运行其Terraform或Ansible工具，生成变更计划，供SRE团队审批。
• 数据工程 Agent： 当Agent负责 ETL（抽取、转换、加载）数据时，Dry-run模式可以模拟数据转换和加载过程，预览最终数据结构和内容，而不会污染生产数据仓库。
• 安全运维 Agent： Agent检测到潜在安全威胁后，可能会建议执行一系列修复操作（如隔离受感染主机、更新防火墙规则、撤销用户权限）。在执行这些高风险操作前，Dry-run模式可以展示每一步操作的影响，避免误伤。
• 内容管理 Agent： 如果Agent负责生成和发布内容（如博客文章、社交媒体更新），Dry-run模式可以预览最终发布的内容格式、链接和图片，避免发布错误信息。
• 客户服务 Agent： 某些高级Agent可以代表用户执行操作（如修改订单、更新账户信息）。Dry-run模式可以向用户展示这些操作的预览，并获得用户确认。

通过在这些关键场景中引入Dry-run Mode，我们能够显著提升Agent系统的可靠性、安全性和用户信任度。

七、 负责任的 Agent 开发之道

Dry-run Mode 是构建负责任、可信赖智能代理系统的基石之一。它在Agent的自主性与系统安全性之间架起了一座桥梁，使得开发者能够在快速迭代和探索Agent能力的同时，依然保有对系统行为的最终控制。通过精心设计和实现Dry-run机制，我们能够让Agent在真实世界中发挥其强大潜力，同时最大限度地降低非预期副作用带来的风险，确保Agent在复杂多变的环境中能够安全、可靠、可预测地运行。这是一个持续演进的领域，对Dry-run模式的投入，是对未来自动化和智能化系统质量的投入。