AI Agents · 课时

在代理环境中读写文件

使用 open()、pathlib.Path,从代理工具读取 CSV/JSON/TXT 文件。

第 1 / 4 课13 个步骤

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

为什么文件输入/输出对代理很重要

代理经常需要读取配置、存储中间结果以及写入输出报告。Python 提供了两种处理文件的主要方式:内置的 open() 函数和现代的 pathlib 模块。两者都必须掌握。

每次文件操作都应使用上下文管理器(with 语句),以确保即使发生错误,文件也会被关闭。

with open('config.txt', 'w', encoding='utf-8') as f:
    f.write('model=gpt-4o-mini\nmax_steps=20\n')

with open('config.txt', 'r', encoding='utf-8') as f:
    content = f.read()

from pathlib import Path
content = Path('config.txt').read_text(encoding='utf-8')

print(content)

使用 open() 读取文件

open() 函数接收文件名和模式。要读取文本,请使用模式 'r'。您可以使用 .read() 一次性读取整个文件,使用 .readline() 逐行读取,或者使用 .readlines() 将所有行读入列表。

请始终指定 encoding='utf-8',以避免平台相关的编码问题。

with open('report.txt', 'w', encoding='utf-8') as f:
    f.write('Quarterly agent report body text.')
with open('large_log.txt', 'w', encoding='utf-8') as f:
    f.write('INFO: started\nERROR: timeout\nINFO: done\n')
with open('tasks.txt', 'w', encoding='utf-8') as f:
    f.write('task one\ntask two\n')

with open('report.txt', 'r', encoding='utf-8') as f:
    full_text = f.read()
    print(f'Read {len(full_text)} characters')

with open('large_log.txt', 'r', encoding='utf-8') as f:
    for line in f:
        line = line.rstrip('\n')
        if 'ERROR' in line:
            print('Found error:', line)

with open('tasks.txt', 'r', encoding='utf-8') as f:
    tasks = f.readlines()
    tasks = [t.strip() for t in tasks]
print('Tasks:', tasks)

使用 open() 写入文件

使用模式 'w' 写入(创建或覆盖),使用 'a' 追加,使用 'x' 独占创建(如果文件已存在则失败)。模式 'w' 会先截断文件——处理现有数据时请务必小心。

# Write a new file (or overwrite)
results = ['Task 1 complete', 'Task 2 complete', 'Task 3 failed']

with open('results.txt', 'w', encoding='utf-8') as f:
    for result in results:
        f.write(result + '\n')

# Append to existing file (add without overwriting)
with open('agent_log.txt', 'a', encoding='utf-8') as f:
    f.write('2026-05-29 14:30: Agent completed batch\n')

# Create exclusively — error if file already exists
try:
    with open('unique_report.txt', 'x', encoding='utf-8') as f:
        f.write('This file is new')
except FileExistsError:
    print('File already exists, skipping')

# --- demo ---
print('results.txt contents:')
with open('results.txt', encoding='utf-8') as f:
    print(f.read())

# Try the exclusive create again now that the file exists
try:
    with open('unique_report.txt', 'x', encoding='utf-8') as f:
        f.write('This file is new')
except FileExistsError:
    print('File already exists, skipping')

pathlib.Path — 现代文件操作

pathlib.Path 为文件路径提供了面向对象的处理方式。与 os.path 相比,它的可读性更好,并且会自动处理跨平台的路径分隔符。对于简单的读写操作,read_text() 和 write_text() 是最简洁的方式。

from pathlib import Path

base = Path('agent_workspace')
base.mkdir(exist_ok=True)
config_file = base / 'config.json'
config_file.write_text('{"model": "gpt-4o-mini"}', encoding='utf-8')

if config_file.exists():
    content = config_file.read_text(encoding='utf-8')
    print('Config:', content)

output = base / 'output' / 'report.txt'
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text('Agent report\nLine 2', encoding='utf-8')

Path('image.png').write_bytes(b'\x89PNG\r\n\x1a\n' + bytes(range(100)))
binary_data = Path('image.png').read_bytes()
print(f'File size: {len(binary_data)} bytes')

读写 JSON 文件

智能体经常通过 JSON 文件交换结构化数据。使用 json.load() 解析 JSON 文件,使用 json.dump() 写入 JSON 文件。为了生成便于人类阅读的输出,请始终使用 indent=2,并使用 ensure_ascii=False 保留 Unicode 字符。

import json
from pathlib import Path

config_path = Path('agent_config.json')
config_path.write_text(json.dumps({'max_retries': 5, 'model': 'gpt-4o-mini'}), encoding='utf-8')

with open(config_path, 'r', encoding='utf-8') as f:
    config = json.load(f)

print('Max retries:', config.get('max_retries', 3))
print('Model:', config.get('model', 'gpt-4o-mini'))

results = {
    'run_id': 'run-001',
    'status': 'completed',
    'items_processed': 142,
    'errors': []
}

with open('results.json', 'w', encoding='utf-8') as f:
    json.dump(results, f, indent=2, ensure_ascii=False)

print('Results written to results.json')

上下文管理器 — 正确用法

with 语句(上下文管理器)可以保证代码块退出时文件已经关闭,即使读取过程中发生异常也不例外。如果不使用它,文件可能会一直处于打开状态并泄漏文件描述符,从而在文件句柄数量受限的系统上引发崩溃。

with open('data.txt', 'w', encoding='utf-8') as f:
    f.write('sample content')

f = open('data.txt', 'r')
try:
    content = f.read()
finally:
    f.close()

with open('data.txt', 'r', encoding='utf-8') as f:
    content = f.read()

with open('input.txt', 'w', encoding='utf-8') as f:
    f.write('hello\nworld\n')

with open('input.txt', 'r') as infile, \
     open('output.txt', 'w') as outfile:
    for line in infile:
        processed = line.upper()
        outfile.write(processed)

print('Done:', content)

处理二进制文件

使用模式 'rb' 读取二进制文件(图像、PDF、音频),使用 'wb' 写入二进制文件。切勿尝试将二进制内容解码为文本;请结合 pathlib 使用 read_bytes()/write_bytes(),或者在 open() 中使用二进制模式。

from pathlib import Path
import base64

image_path = Path('screenshot.png')
image_path.write_bytes(b'\x89PNG\r\n\x1a\n' + bytes(range(50)))

image_bytes = image_path.read_bytes()
print(f'Image size: {len(image_bytes)} bytes')

image_b64 = base64.b64encode(image_bytes).decode('utf-8')
print(f'Base64 length: {len(image_b64)}')

response_bytes = b'...binary content from API...'
output_path = Path('output') / 'generated_image.png'
output_path.parent.mkdir(exist_ok=True)
output_path.write_bytes(response_bytes)
print('Wrote', output_path)

处理编码和换行问题

不同系统使用不同的换行符(Linux/macOS 上使用 \n,Windows 上使用 \r\n)。请始终指定 encoding='utf-8',并使用 newline=None(默认值)让 Python 处理换行符。对于包含 BOM(字节顺序标记)的文件,请使用 encoding='utf-8-sig'。

with open('data.txt', 'w', encoding='utf-8') as f:
    f.write('hello world')
with open('windows_export.csv', 'wb') as f:
    f.write('name,email\n'.encode('utf-8'))
with open('unknown_encoding.txt', 'wb') as f:
    f.write('cafe resume'.encode('latin-1'))

with open('data.txt', 'r', encoding='utf-8') as f:
    text = f.read()

with open('windows_export.csv', 'r', encoding='utf-8-sig') as f:
    content = f.read()

try:
    with open('unknown_encoding.txt', 'r', encoding='utf-8') as f:
        text = f.read()
except UnicodeDecodeError:
    with open('unknown_encoding.txt', 'r',
               encoding='utf-8', errors='replace') as f:
        text = f.read()
    print('Warning: file had encoding errors (replaced with ?)')

print('All reads completed. csv content:', repr(content))

高效地逐行处理

对于大型文件(日志文件、数据集),请避免使用 .read() 将所有内容加载到内存中。直接遍历文件对象即可;Python 会一次读取一行,因此无论文件多大,内存使用量都能保持不变。

from pathlib import Path

def process_large_log(log_path):
    error_count = 0
    processed = 0

    with open(log_path, 'r', encoding='utf-8') as f:
        for line in f:  # reads one line at a time
            line = line.strip()
            if not line:
                continue

            processed += 1
            if '[ERROR]' in line:
                error_count += 1
                # Process error immediately without storing all lines
                handle_error(line)

    print(f'Processed {processed} lines, {error_count} errors')

def handle_error(line):
    print(f'Error found: {line[:80]}')

# --- demo ---
with open('demo_agent.log', 'w', encoding='utf-8') as f:
    f.write('[INFO] agent started\n')
    f.write('[ERROR] timeout calling tool search_web\n')
    f.write('[INFO] retrying\n')
    f.write('[ERROR] tool search_web failed again\n')

process_large_log('demo_agent.log')

创建和管理目录

写入文件之前,请确保父目录存在。使用 Path.mkdir(parents=True, exist_ok=True) 创建嵌套目录;如果目录已经存在,也不会产生错误。这是任何会生成输出文件的智能体在开始运行时常见的准备步骤。

from pathlib import Path
import datetime

# Create output directory structure
today = datetime.date.today().isoformat()  # '2026-05-29'
output_dir = Path('agent_output') / today / 'reports'
output_dir.mkdir(parents=True, exist_ok=True)

print(f'Output directory: {output_dir}')

# Now write files safely
for i in range(3):
    report_path = output_dir / f'report_{i:02d}.json'
    report_path.write_text(
        '{"status": "ok"}',
        encoding='utf-8'
    )

# List all created files
for f in output_dir.iterdir():
    print(f'  {f.name}: {f.stat().st_size} bytes')

智能体的文件路径最佳实践

智能体应使用绝对路径,或使用相对于明确定义的基准目录的路径。切勿依赖当前工作目录(./),因为智能体经常从不同位置启动。请使用 Path(__file__).parent 获取脚本所在目录,并将其作为基准目录。

from pathlib import Path

# Get absolute path relative to THIS script's location
SCRIPT_DIR = Path(__file__).parent
DATA_DIR = SCRIPT_DIR / 'data'
OUTPUT_DIR = SCRIPT_DIR / 'output'

# Always resolve to absolute path before use
config = DATA_DIR / 'config.json'
print('Config path:', config.resolve())

# Check if file exists before reading
if config.exists():
    import json
    with open(config, 'r', encoding='utf-8') as f:
        settings = json.load(f)
else:
    print(f'Config not found at {config.resolve()}')
    settings = {}

# Safe output path
result_path = (OUTPUT_DIR / 'result.json').resolve()
print('Will write to:', result_path)

快速检查:上下文管理器

检验您对文件输入/输出最佳实践的理解。

文件输入/输出回顾

现在,您已经掌握了智能体代码中文件操作的坚实基础:

  • 使用 open('file', 'r', encoding='utf-8') 读取文件,使用 'w' 写入文件,使用 'a' 追加内容
  • 始终使用 with 语句,确保文件已关闭
  • pathlib.Path 提供简洁且跨平台的路径处理方式,并支持 read_text()/write_text()
  • 对于结构化数据,使用 json.load(f) / json.dump(data, f, indent=2)
  • 处理大型文件时逐行遍历,以尽量减少内存使用
  • 使用 Path(__file__).parent 作为基准目录 — 切勿依赖 cwd
免费开始

用 AI 导师学习 AI Agents — 免费

在浏览器中编写并运行真实代码,获得全天候 AI 导师的即时帮助,并在网页或应用中继续学习。

课程
60
课程
239

常见问题解答

「在代理环境中读写文件」课时是免费的吗?

是的 — 「在代理环境中读写文件」的完整文本可在网页上免费阅读。要进行交互式练习(内置代码编辑器和全天候 AI 导师)并解锁 AI Agents 课程的其余内容,请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「在代理环境中读写文件」这节课中我会学到什么?

使用 open()、pathlib.Path,从代理工具读取 CSV/JSON/TXT 文件。 你通过在浏览器中直接运行的动手代码来练习 AI Agents,全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗?

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

「在代理环境中读写文件」课时需要多长时间?

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

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

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

此课程中的所有课时

  1. 在代理环境中读写文件
  2. 目录遍历与文件发现
  3. 文件格式处理:CSV、JSON 与 TXT
  4. 带错误处理的安全文件操作
← 返回 AI Agents