在代理环境中读写文件
使用 open()、pathlib.Path,从代理工具读取 CSV/JSON/TXT 文件。
在代理环境中读写文件 是 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 反馈 — 无需本地设置。