4. 代码执行与环境

本节导读：掌握AutoGen代码执行机制，学习Docker环境配置、安全控制优化和执行环境管理，确保智能体代码安全可靠运行

学习目标

• 理解AutoGen代码执行的核心机制和原理
• 掌握Docker环境的配置和管理方法
• 学会实现安全的代码执行策略
• 了解不同执行器的适用场景和配置要点

核心概念

代码执行是AutoGen框架的重要特性，允许智能体生成并执行Python代码。通过安全可控的执行环境，实现从代码生成到运行验证的完整流程。

环境准备 / 前置知识

• Python 3.10+ 环境
• Docker基础知识和安装
• 基础Linux命令操作
• 网络和安全概念

执行器配置

LocalCommandLineCodeExecutor

本地命令行执行器，在本地环境执行代码：

from autogen.code_executor import LocalCommandLineCodeExecutor

# 创建本地执行器
executor = LocalCommandLineCodeExecutor(
    timeout=300,  # 超时时间（秒）
    work_dir="/tmp/autogen_work",  # 工作目录
    execution_timeout=30  # 单次执行超时
)

# 使用执行器
result = executor.execute_code(
    code="print('Hello, AutoGen!')",
    filename="test.py"
)
​

DockerCommandLineCodeExecutor

Docker容器执行器，提供隔离的执行环境：

from autogen.code_executor import DockerCommandLineCodeExecutor

# 创建Docker执行器
executor = DockerCommandLineCodeExecutor(
    image="python:3.10-slim",  # 基础镜像
    timeout=300,  # 超时时间
    work_dir="/workspace",  # 容器内工作目录
    container_name="autogen-executor"  # 容器名称
)

# 使用执行器
result = executor.execute_code(
    code="""
import pandas as pd
import numpy as np

# 创建示例数据
data = pd.DataFrame({
    'A': np.random.randn(100),
    'B': np.random.randn(100)
})
print(data.head())
""",
    filename="data_analysis.py"
)
​

自定义执行器

可以继承基础执行器类创建自定义执行器：

from autogen.code_executor import CodeExecutor

class CustomCodeExecutor(CodeExecutor):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.custom_dependencies = []
    
    def setup_environment(self, code: str) -> str:
        """自定义环境设置"""
        setup_script = ""
        for dep in self.custom_dependencies:
            setup_script += f"pip install {dep}\n"
        return setup_script + code
    
    def execute_code(self, code: str, filename: str = None) -> dict:
        """自定义执行逻辑"""
        full_code = self.setup_environment(code)
        return super().execute_code(full_code, filename)
​

Docker集成

Docker环境配置

# Docker配置文件
docker_config = {
    "base_image": "python:3.10-slim",
    "packages": [
        "pandas>=1.5.0",
        "numpy>=1.21.0",
        "matplotlib>=3.5.0"
    ],
    "environment_vars": {
        "PYTHONPATH": "/app",
        "DATA_DIR": "/data"
    },
    "volume_mounts": {
        "/host/data": "/container/data"
    }
}
​

Dockerfile定制

FROM python:3.10-slim

# 安装系统依赖
RUN apt-get update && apt-get install -y \
    git \
    curl \
    && rm -rf /var/lib/apt/lists/*

# 设置工作目录
WORKDIR /workspace

# 安装Python依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 设置环境变量
ENV PYTHONPATH=/workspace
ENV DATA_DIR=/data

# 启动命令
CMD ["python", "-m", "autogen.executor"]
​

Docker Compose编排

version: '3.8'
services:
  autogen-executor:
    build: .
    ports:
      - "8080:8080"
    volumes:
      - ./data:/workspace/data
      - ./code:/workspace/code
    environment:
      - PYTHONPATH=/workspace
      - DATA_DIR=/workspace/data
    restart: unless-stopped
​

安全控制

执行策略

class SecureCodeExecutor:
    def __init__(self):
        self.allowed_modules = {
            'pandas', 'numpy', 'matplotlib', 'requests'
        }
        self.blocked_imports = {
            'os', 'subprocess', 'sys', 'socket'
        }
    
    def validate_code(self, code: str) -> bool:
        """验证代码安全性"""
        # 检查导入模块
        imports = self.extract_imports(code)
        for module in imports:
            if module in self.blocked_imports:
                return False
            if module not in self.allowed_modules:
                # 可以动态加载模块
                if not self.is_module_safe(module):
                    return False
        return True
    
    def extract_imports(self, code: str) -> set:
        """提取代码中的导入语句"""
        # 实现导入提取逻辑
        pass
    
    def is_module_safe(self, module: str) -> bool:
        """检查模块安全性"""
        # 实现模块安全性检查
        pass
​

资源限制

import resource
import signal

class ResourceLimitedExecutor:
    def __init__(self, max_memory_mb=512, max_cpu_time=30):
        self.max_memory = max_memory_mb * 1024 * 1024
        self.max_cpu_time = max_cpu_time
    
    def execute_with_limits(self, code: str):
        """在资源限制下执行代码"""
        def resource_limit_handler(signum, frame):
            raise MemoryError("Memory limit exceeded")
        
        # 设置内存限制
        resource.setrlimit(resource.RLIMIT_AS, (self.max_memory, self.max_memory))
        
        # 设置CPU时间限制
        signal.signal(signal.SIGXCPU, resource_limit_handler)
        resource.setrlimit(resource.RLIMIT_CPU, (self.max_cpu_time, self.max_cpu_time))
        
        try:
            exec(code)
        except MemoryError:
            raise
        finally:
            # 恢复限制
            resource.setrlimit(resource.RLIMIT_AS, (resource.RLIM_INFINITY, resource.RLIM_INFINITY))
            resource.setrlimit(resource.RLIMIT_CPU, (resource.RLIM_INFINITY, resource.RLIM_INFINITY))
​

沙盒环境

import tempfile
import os
import sys
from contextlib import contextmanager

@contextmanager
def sandbox_environment():
    """创建代码执行的沙盒环境"""
    # 创建临时目录
    with tempfile.TemporaryDirectory() as temp_dir:
        # 设置工作目录
        original_cwd = os.getcwd()
        os.chdir(temp_dir)
        
        # 限制导入模块
        original_path = sys.path[:]
        sys.path = [temp_dir]
        
        try:
            yield temp_dir
        finally:
            # 恢复环境
            os.chdir(original_cwd)
            sys.path = original_path
            
            # 清理临时文件
            for item in os.listdir(temp_dir):
                item_path = os.path.join(temp_dir, item)
                if os.path.isfile(item_path):
                    os.unlink(item_path)
​

完整示例

多环境执行系统

import asyncio
from autogen.code_executor import DockerCommandLineCodeExecutor
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient

class MultiEnvironmentExecutor:
    def __init__(self):
        # 定义不同环境的执行器
        self.executors = {
            "local": LocalCommandLineCodeExecutor(
                timeout=60, work_dir="/tmp/local"
            ),
            "docker": DockerCommandLineCodeExecutor(
                image="python:3.10-slim",
                timeout=300,
                work_dir="/workspace"
            ),
            "jupyter": JupyterCodeExecutor(
                timeout=120,
                kernel_name="python3"
            )
        }
    
    async def execute_code_analysis(self, code: str, environment: str = "docker"):
        """在指定环境中执行代码分析"""
        if environment not in self.executors:
            raise ValueError(f"不支持的执行环境: {environment}")
        
        executor = self.executors[environment]
        
        # 代码安全检查
        if not self.validate_code_safety(code):
            raise SecurityError("代码安全性检查失败")
        
        # 执行代码
        result = executor.execute_code(code, "analysis.py")
        
        # 分析执行结果
        analysis = await self.analyze_execution_result(result)
        
        return {
            "execution_result": result,
            "analysis": analysis,
            "environment": environment
        }
    
    def validate_code_safety(self, code: str) -> bool:
        """验证代码安全性"""
        # 实现安全检查逻辑
        return True
    
    async def analyze_execution_result(self, result: dict):
        """分析执行结果"""
        # 实现结果分析逻辑
        pass

# 使用示例
async def main():
    executor = MultiEnvironmentExecutor()
    
    # 分析代码
    analysis_code = """
import pandas as pd
import matplotlib.pyplot as plt

# 生成示例数据
data = pd.DataFrame({
    'x': range(100),
    'y': [x**2 for x in range(100)]
})

# 创建图表
plt.figure(figsize=(10, 6))
plt.plot(data['x'], data['y'])
plt.title('Sample Data Analysis')
plt.xlabel('X values')
plt.ylabel('Y values')
plt.grid(True)
plt.savefig('analysis.png')
plt.close()

print("Analysis completed successfully")
"""
    
    result = await executor.execute_code_analysis(analysis_code, "docker")
    print("Code analysis result:", result)

asyncio.run(main())
​

常见问题 FAQ

Q1：如何处理长时间运行的代码？

A：设置合理的超时时间，实现异步执行和超时控制机制。

Q2：Docker容器如何处理网络访问？

A：通过配置网络模式、端口映射和网络策略来控制网络访问。

Q3：如何保证代码执行的性能？

A：优化资源配置、使用缓存机制、合理设置并发数和批量处理策略。

最佳实践与避坑

• 环境隔离：始终使用隔离的执行环境，避免系统污染
• 资源监控：实现资源使用监控和限制
• 日志记录：完整的执行日志记录，便于调试和审计
• 版本控制：使用版本化的基础镜像和环境配置

本节小结

本节详细介绍了AutoGen代码执行的机制、Docker环境配置和安全控制策略。通过实际示例，读者可以理解如何构建安全、高效的代码执行环境。下一节将探讨扩展与集成的实践方法。

延伸阅读

• Docker官方文档
• Python代码执行安全指南
• 容器化最佳实践

关键词：代码执行, Docker环境, 安全控制, 执行器配置, 沙盒机制
难度：进阶
预计阅读：40 分钟