0Pricing
AI Agents · 课时

用于数据分析的代码解释器模式

沙箱化的 Python 执行：在代理工具中运行 pandas/matplotlib。

用于数据分析的代码解释器模式 是 CoddyKit 上的免费 AI Agents 课时。 这是第 1 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

代码解释器模式

代码解释器模式允许智能体生成 Python 代码来回答数据分析问题，在沙箱中执行代码，捕获输出，并解释结果。

智能体不必为每种分析操作都预先编写代码，而是针对每个问题编写自定义代码，因此能够灵活处理各种数据任务。

核心循环：生成 → 执行 → 解释

这一模式包含三个可以重复执行的步骤：

  1. 生成 — LLM 编写用于回答问题的 Python 代码
  2. 执行 — 在沙箱中运行代码，并捕获标准输出和文件
  3. 解释 — 将输出传回 LLM，由它解释结果
def code_interpreter_agent(question, data_path):
    # Step 1: Generate code
    code = generate_analysis_code(question, data_path)
    print('Generated code:', code[:200])

    # Step 2: Execute in sandbox
    result = execute_in_sandbox(code)

    if result['error']:
        # Try to fix the error
        fixed_code = fix_code(code, result['error'])
        result = execute_in_sandbox(fixed_code)

    # Step 3: Interpret output
    return interpret_output(question, result)

print(code_interpreter_agent(
    'What is the average order value by customer segment?',
    'data/orders.csv'
))

代码生成提示词

代码生成提示词必须包含：数据路径或数据模式、问题，以及约束条件（不得调用外部 API、使用 pandas、将图表保存为文件）。

CODE_GEN_PROMPT = """You are a Python data analyst. Write Python code to answer the question.
Data available at: {data_path}
Question: {question}
Write ONLY Python code (no markdown, no explanation):"""

def llm_call(prompt):
    return "```python\nprint('df.describe() results')\n```"

def generate_analysis_code(question, data_path):
    response = llm_call(CODE_GEN_PROMPT.format(question=question, data_path=data_path))
    code = response.strip()
    if code.startswith('```'):
        code = code.split('```')[1]
        if code.startswith('python'):
            code = code[6:]
    return code.strip()

print(generate_analysis_code('What is the average price?', 'data.csv'))

子进程沙箱执行

最简单的沙箱，是在带有超时限制的独立子进程中运行代码。这样可以实现进程隔离——即使代码崩溃，也不会拖垮代理。

import subprocess
import tempfile
import os

def execute_in_sandbox(code, timeout=30):
    # Write code to temp file
    with tempfile.NamedTemporaryFile(suffix='.py', mode='w', delete=False) as f:
        f.write(code)
        script_path = f.name

    try:
        result = subprocess.run(
            ['python3', script_path],
            capture_output=True,
            text=True,
            timeout=timeout,
            env={**os.environ, 'MPLBACKEND': 'Agg'}  # non-interactive matplotlib
        )
        return {
            'stdout': result.stdout,
            'stderr': result.stderr,
            'returncode': result.returncode,
            'error': result.stderr if result.returncode != 0 else None
        }
    except subprocess.TimeoutExpired:
        return {'stdout': '', 'stderr': 'Timeout', 'returncode': -1, 'error': 'Code timed out'}
    finally:
        os.unlink(script_path)

if __name__ == '__main__':
    result = execute_in_sandbox('print(2 + 2)')
    print('Sandbox stdout:', result['stdout'].strip())
    print('Return code   :', result['returncode'])

E2B 云端沙箱

E2B 提供用于安全执行代码的托管云端沙箱——比子进程更加安全。它会在隔离容器中运行代码，并提供文件系统访问权限。

使用 pip install e2b-code-interpreter 安装。

from e2b_code_interpreter import Sandbox
import os

def execute_with_e2b(code, data_bytes=None):
    with Sandbox(api_key=os.getenv('E2B_API_KEY')) as sandbox:
        # Upload data file if provided
        if data_bytes:
            sandbox.files.write('/home/user/data.csv', data_bytes)

        # Execute code
        execution = sandbox.run_code(code)

        result = {
            'stdout': '\n'.join(execution.logs.stdout),
            'stderr': '\n'.join(execution.logs.stderr),
            'error': None
        }

        # Check for errors
        if execution.error:
            result['error'] = str(execution.error)

        # Download any generated files
        result['files'] = []
        for output in execution.results:
            if hasattr(output, 'png'):
                result['files'].append({
                    'type': 'image/png',
                    'data': output.png  # base64 encoded
                })

    return result

捕获生成的文件

代码可能会生成图表、CSV 导出文件或其他文件。请从沙箱文件系统中捕获这些文件，并将它们传回代理，以便进行解读或显示。

import os
import glob
import base64

OUTPUT_DIR = '/tmp/chart_output'

def execute_and_capture(code, timeout=30):
    # Create output dir
    os.makedirs(OUTPUT_DIR, exist_ok=True)

    result = execute_in_sandbox(code, timeout=timeout)

    # Capture any generated image files
    generated_files = []
    for filepath in glob.glob(os.path.join(OUTPUT_DIR, '*.png')):
        with open(filepath, 'rb') as f:
            encoded = base64.b64encode(f.read()).decode('utf-8')
        generated_files.append({
            'filename': os.path.basename(filepath),
            'type': 'image/png',
            'base64': encoded
        })
        os.unlink(filepath)  # clean up

    result['generated_files'] = generated_files
    print(f'Captured {len(generated_files)} file(s) from sandbox')
    return result

错误恢复循环

生成的代码首次运行时经常会出现错误。请实现一个恢复循环：将错误连同原始代码发回 LLM，并请求修复。将重试次数限制为 2～3 次。

FIX_PROMPT = '''The following Python code raised an error. Fix it.

Original code:
{code}

Error:
{error}

Return ONLY the fixed Python code (no explanation, no markdown):'''

def fix_code(code, error):
    return llm_call(FIX_PROMPT.format(code=code, error=error)).strip()

def execute_with_retry(code, max_retries=2):
    for attempt in range(max_retries + 1):
        result = execute_and_capture(code)
        if not result['error']:
            return result
        print(f'Attempt {attempt + 1} failed: {result["error"][:100]}')
        if attempt < max_retries:
            code = fix_code(code, result['error'])
    return result  # return last result even if errored

解读代码输出

原始代码输出（数字、表格）需要转换为自然语言答案。将标准输出传给 LLM，并要求它结合原始问题解释结果。

INTERPRET_PROMPT = '''A Python script was executed to answer a data analysis question.
Explain the results in clear, non-technical language.

Original question: {question}

Code output (stdout):
{output}

Provide a clear, concise answer that directly addresses the question.
Highlight the most important numbers or findings.
Answer:'''

def interpret_output(question, execution_result):
    stdout = execution_result.get('stdout', '').strip()
    error = execution_result.get('error')

    if error and not stdout:
        return f'The analysis failed with error: {error}'

    if not stdout:
        return 'The code ran successfully but produced no output.'

    return llm_call(INTERPRET_PROMPT.format(
        question=question,
        output=stdout[:3000]  # truncate very long outputs
    ))

代码生成中的安全约束

生成的代码不得发起网络请求、访问敏感文件或执行系统命令。请同时通过提示词和沙箱限制来强制执行这些要求。

BLOCKED_IMPORTS = ['requests', 'httpx', 'urllib', 'socket', 'subprocess', 'os.system']

def pre_validate_code(code):
    errors = []
    for blocked in BLOCKED_IMPORTS:
        if f'import {blocked}' in code or f'from {blocked}' in code:
            errors.append(f'Blocked import: {blocked}')

    # Block shell execution
    import re
    if re.search(r'os\.system|subprocess\.run|subprocess\.call|eval\(|exec\(', code):
        errors.append('Blocked: shell execution or eval/exec')

    # Block reading outside allowed paths
    if re.search(r'open\([^)]*\.\./|open\([^)]*\/etc\/', code):
        errors.append('Blocked: unauthorized file access')

    if errors:
        raise ValueError('Security check failed:\n' + '\n'.join(errors))

    return True

if __name__ == '__main__':
    try:
        pre_validate_code('import requests\nrequests.get("http://x")')
    except ValueError as e:
        print('Rejected:', e)
    print('Safe code passed:', pre_validate_code('print(1 + 1)'))

注入数据模式

如果 LLM 预先了解数据模式，就能生成更好的代码——包括列名、类型和示例行。在代码生成提示词中加入数据模式描述。

import pandas as pd

def get_data_schema(data_path):
    df = pd.read_csv(data_path, nrows=5)
    schema_lines = []
    for col in df.columns:
        dtype = str(df[col].dtype)
        sample = df[col].dropna().iloc[0] if len(df[col].dropna()) > 0 else 'N/A'
        schema_lines.append(f'  - {col} ({dtype}): sample={sample!r}')
    schema_text = '\n'.join(schema_lines)
    return f'CSV columns:\n{schema_text}\nTotal rows: {len(pd.read_csv(data_path))}'

ENHANCED_PROMPT = CODE_GEN_PROMPT + '\n\nData schema:\n{schema}'

def generate_analysis_code_with_schema(question, data_path):
    schema = get_data_schema(data_path)
    response = llm_call(ENHANCED_PROMPT.format(
        question=question, data_path=data_path, schema=schema
    ))
    return response.strip()

跟踪执行历史

保留会话中所有代码执行的日志。这样代理就能引用之前的结果、基于早先的计算继续处理，并向用户解释分析步骤。

from datetime import datetime

execution_history = []

def record_execution(question, code, result):
    execution_history.append({
        'timestamp': datetime.now().isoformat(),
        'question': question,
        'code_lines': len(code.splitlines()),
        'stdout_preview': result.get('stdout', '')[:200],
        'success': result.get('error') is None,
        'generated_files': len(result.get('generated_files', []))
    })

def get_session_summary():
    total = len(execution_history)
    successful = sum(1 for e in execution_history if e['success'])
    return {
        'total_executions': total,
        'successful': successful,
        'failed': total - successful,
        'success_rate': f'{successful/max(total,1)*100:.0f}%',
        'questions_answered': [e['question'][:60] for e in execution_history]
    }

# Usage in agent loop
def code_interpreter_agent_tracked(question, data_path):
    code = generate_analysis_code_with_schema(question, data_path)
    result = execute_with_retry(code)
    record_execution(question, code, result)
    return interpret_output(question, result)

知识检查

相较于将特定数据分析操作硬编码为代理工具，使用代码解释器模式的主要优势是什么？

回顾：用于数据分析的代码解释器模式

代码解释器模式的流程是：生成 Python 代码（包含数据模式上下文和安全约束）→ 在沙箱中执行（子进程或 E2B）→ 捕获标准输出和文件→ 发生错误时重试→ 用自然语言解读结果。

关键注意事项：注入数据模式以改进代码生成；预先验证代码，检查被禁止的导入和 Shell 命令；使用子进程超时限制，防止执行失控；并将生成的图表捕获为 base64，以便显示。

常见问题解答

「用于数据分析的代码解释器模式」课时是免费的吗？

是的 — 「用于数据分析的代码解释器模式」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「用于数据分析的代码解释器模式」这节课中我会学到什么？

沙箱化的 Python 执行：在代理工具中运行 pandas/matplotlib。 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 1 节课，共 4 节。

「用于数据分析的代码解释器模式」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 用于数据分析的代码解释器模式
  2. 由 Pandas 驱动的数据代理工具
  3. 自动生成图表与可视化内容
  4. 统计摘要代理
← 返回 AI Agents