文件格式处理：CSV、JSON 与 TXT
csv 模块、json.load/dump，以及适用于代理工具的安全文本编码。
文件格式处理：CSV、JSON 与 TXT 是 CoddyKit 上的免费 AI Agents 课时。 这是第 3 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。
智能体最常用的三种文件格式
智能体经常读写三种文件格式：用于表格数据的 CSV、用于结构化对象的 JSON，以及用于日志、提示词和报告的纯文本。每种格式都有不同的解析要求、边界情况和最佳实践。Python 对这三种格式都提供了出色的内置支持。
import csv
import json
from pathlib import Path
# Detect format from extension
def read_data_file(file_path):
path = Path(file_path)
if path.suffix == '.csv':
return read_csv(path)
elif path.suffix == '.json':
return read_json(path)
elif path.suffix == '.txt':
return path.read_text(encoding='utf-8')
else:
raise ValueError(f'Unsupported format: {path.suffix}')csv.reader — 基本 CSV 解析
csv.reader 会逐行解析 CSV 文件，并将每一行作为字符串列表返回。它能够正确处理带引号的字段、字段中的逗号，以及带引号值中的换行符；而手动按逗号拆分会在这些边界情况下出错。
import csv
with open('sales.csv', 'w', newline='') as f:
f.write('product,qty,price\nWidget,3,9.99\nGadget,1,19.99\n')
with open('sales.csv', 'r', encoding='utf-8', newline='') as f:
reader = csv.reader(f)
header = next(reader)
print('Columns:', header)
for row in reader:
product = row[0]
quantity = int(row[1])
price = float(row[2])
print(f'{product}: {quantity} units at ${price}')csv.DictReader — 将行读取为字典
csv.DictReader 会将每一行读取为一个 OrderedDict（在 Python 3.8 及更高版本中为普通字典），并将列标题作为键。与按位置索引相比，这种方式更易于处理 — 即使列的顺序发生变化，代码仍然清晰易读。
import csv
with open('employees.csv', 'w', newline='') as f:
f.write('name,department,salary\nAlice,Eng,95000\nBob,Sales,70000\n')
with open('employees.csv', 'r', encoding='utf-8', newline='') as f:
reader = csv.DictReader(f)
print('Fields:', reader.fieldnames)
total_salary = 0
for row in reader:
name = row['name']
department = row['department']
salary = float(row['salary'])
total_salary += salary
print(f'{name} ({department}): ${salary:,.2f}')
print(f'Total payroll: ${total_salary:,.2f}')csv.writer 和 DictWriter — 写入 CSV
使用 csv.writer 将列表形式的行写入文件，或使用 csv.DictWriter 将字典形式的行写入文件。打开文件时请始终传入 newline='' — csv 模块会自行处理行尾，从而避免 Windows 上出现重复换行。
import csv
results = [
{'task_id': 'T001', 'status': 'completed', 'duration_s': 12.5},
{'task_id': 'T002', 'status': 'failed', 'duration_s': 3.1},
{'task_id': 'T003', 'status': 'completed', 'duration_s': 45.8},
]
fieldnames = ['task_id', 'status', 'duration_s']
with open('task_results.csv', 'w', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader() # write column names
writer.writerows(results)
print('Wrote task_results.csv')json.load() 和 json.dump() — 文件输入/输出
使用 json.load(file) 解析 JSON 文件，使用 json.dump(obj, file) 写入 JSON 文件。这些函数处理的是文件对象。对于字符串，请使用 json.loads(string) 和 json.dumps(obj)。为了生成易读的输出，请始终使用 indent=2。
import json
with open('config.json', 'w', encoding='utf-8') as f:
json.dump({'api_url': 'https://api.example.com', 'timeout': 15}, f)
with open('config.json', 'r', encoding='utf-8') as f:
config = json.load(f)
print('API URL:', config.get('api_url'))
print('Timeout:', config.get('timeout', 30))
output_data = {
'run_id': 'abc123',
'items': [1, 2, 3],
'meta': {'agent': 'v2', 'model': 'gpt-4o'}
}
with open('output.json', 'w', encoding='utf-8') as f:
json.dump(
output_data, f,
indent=2,
ensure_ascii=False
)
print('Written output.json')处理 JSON 解码错误
格式错误的 JSON 文件在智能体流水线中很常见，原因可能是写入未完成、下载被截断或编码出现问题。请始终将 json.load() 放在异常处理结构中，并提供包含文件路径的清晰错误消息，以便调试。
import json
from pathlib import Path
def safe_load_json(file_path):
path = Path(file_path)
try:
with open(path, 'r', encoding='utf-8') as f:
return json.load(f)
except json.JSONDecodeError as e:
print(f'Invalid JSON in {path}: line {e.lineno}, col {e.colno}')
print(f' Error: {e.msg}')
# Show the problem area
content = path.read_text(encoding='utf-8')
lines = content.split('\n')
if e.lineno <= len(lines):
print(f' Content: {lines[e.lineno-1][:80]}')
return None
except FileNotFoundError:
print(f'File not found: {path}')
return None
# --- demo ---
Path('good.json').write_text('{"model": "gpt-4o"}', encoding='utf-8')
Path('bad.json').write_text('{"model": "gpt-4o", }', encoding='utf-8')
print('Loading good.json:', safe_load_json('good.json'))
print('Loading bad.json:', safe_load_json('bad.json'))
读取 JSONL（JSON 行）格式
许多 AI API 和数据流水线使用 JSONL（JSON Lines）格式，即每行包含一个 JSON 对象。这种格式支持流式处理，可以逐行处理，而不必将整个文件加载到内存中。每一行都是一个完整且自包含的 JSON 对象。
import json
with open('events.jsonl', 'w', encoding='utf-8') as f:
f.write('{"event": "start"}\n{"event": "stop"}\nnot json\n')
results = []
with open('events.jsonl', 'r', encoding='utf-8') as f:
for line_num, line in enumerate(f, 1):
line = line.strip()
if not line:
continue
try:
event = json.loads(line)
results.append(event)
except json.JSONDecodeError as e:
print(f'Bad JSON on line {line_num}: {e}')
print(f'Loaded {len(results)} events')
with open('output.jsonl', 'w', encoding='utf-8') as f:
for record in results:
f.write(json.dumps(record, ensure_ascii=False) + '\n')读取纯文本文件
纯文本是最简单的格式，可用于日志、提示词、报告和配置文件。您可以使用 .read() 读取整个文件，也可以逐行处理。对于大型文件，请始终采用逐行处理方式，以保持内存使用量不变。
from pathlib import Path
Path('system_prompt.txt').write_text('You are a helpful agent.', encoding='utf-8')
with open('agent.log', 'w', encoding='utf-8') as f:
f.write('INFO: boot\nERROR: disk full\nCRITICAL: crash\nINFO: recovered\n')
prompt = Path('system_prompt.txt').read_text(encoding='utf-8')
print(f'Prompt length: {len(prompt)} characters')
error_lines = []
with open('agent.log', 'r', encoding='utf-8') as f:
for line in f:
line = line.rstrip()
if not line:
continue
if 'ERROR' in line or 'CRITICAL' in line:
error_lines.append(line)
print(f'Found {len(error_lines)} error lines')
with open('summary.txt', 'w', encoding='utf-8') as f:
f.write('Agent Run Summary\n')
f.write('=' * 40 + '\n')
for error in error_lines[:10]:
f.write(f' {error}\n')
print('Summary written')处理 BOM（字节顺序标记）
从 Excel 或 Windows 工具导出的文件通常以 BOM（字节顺序标记）开头，其中包含一个不可见的 \ufeff 字符。如果不处理它，CSV 解析时第一个字段名会被破坏。使用 encoding='utf-8-sig' 可以自动去除该标记。
import csv
with open('windows_export.csv', 'wb') as f:
f.write('name,email,age\nAlice,alice@x.com,30\n'.encode('utf-8'))
with open('file.txt', 'wb') as f:
f.write(b'\xef\xbb\xbfhello')
with open('windows_export.csv', 'r', encoding='utf-8') as f:
reader = csv.DictReader(f)
first = next(reader)
print(list(first.keys()))
with open('windows_export.csv', 'r', encoding='utf-8-sig') as f:
reader = csv.DictReader(f)
first = next(reader)
print(list(first.keys()))
content = open('file.txt', 'rb').read()
if content.startswith(b'\xef\xbb\xbf'):
content = content[3:]
text = content.decode('utf-8')
print('Decoded:', text)处理编码错误
读取来源未知的文件时，编码错误很常见。open() 的 errors 参数控制遇到错误时的处理方式：'replace' 会用 ? 替换错误字符，'ignore' 会丢弃这些字符，'backslashreplace' 会对它们进行转义。若要进行严格验证，请使用 'strict'（默认值）。
from pathlib import Path
def read_with_fallback(file_path):
path = Path(file_path)
# Try UTF-8 first
try:
return path.read_text(encoding='utf-8')
except UnicodeDecodeError:
pass
# Try Latin-1 (handles most European files)
try:
return path.read_text(encoding='latin-1')
except UnicodeDecodeError:
pass
# Last resort: replace bad characters
text = path.read_text(encoding='utf-8', errors='replace')
print(f'Warning: {path.name} had encoding errors (chars replaced)')
return text
# --- demo ---
Path('notes_utf8.txt').write_text('Notes: café, naïve, résumé', encoding='utf-8')
text = read_with_fallback('notes_utf8.txt')
print(f'Read {len(text)} chars: {text!r}')
处理格式错误的 CSV 文件
现实中的 CSV 文件经常存在问题：多余的逗号、缺失的字段、引号使用不一致，或分隔符混用。请使用 quoting 和 error_bad_lines 选项，并将行解析放在异常处理结构中，以便平稳地跳过错误行。
import csv
with open('messy_data.csv', 'w', newline='') as f:
f.write('name,email,score\nAlice,alice@x.com,88\nBob,,90\nCarol,carol@x.com,notanumber\n')
valid_rows = []
error_count = 0
with open('messy_data.csv', 'r', encoding='utf-8', newline='') as f:
reader = csv.DictReader(f)
expected_fields = {'name', 'email', 'score'}
for line_num, row in enumerate(reader, start=2):
try:
if not all(row.get(f, '').strip() for f in expected_fields):
raise ValueError(f'Missing required field in row {line_num}')
score = float(row['score'])
valid_rows.append({
'name': row['name'].strip(),
'email': row['email'].strip().lower(),
'score': score
})
except (ValueError, KeyError) as e:
error_count += 1
print(f'Skipping row {line_num}: {e}')
print(f'Valid: {len(valid_rows)}, Errors: {error_count}')快速检查：CSV DictReader 与 reader
检验您对 CSV 解析选项的理解。
文件格式处理回顾
现在，您已经能够解析和写入智能体使用的主要文件格式：
- CSV：使用
csv.DictReader读取字典形式的行，使用csv.DictWriter写入输出；打开文件时始终使用newline='' - JSON：使用
json.load(f)解析，使用json.dump(obj, f, indent=2)写入；捕获JSONDecodeError - JSONL：使用
json.loads(line)逐行读取并解析；非常适合流式数据 - 纯文本：小型文件使用
.read()，大型文件逐行遍历 - BOM：对于从 Windows 导出的文件，使用
encoding='utf-8-sig' - 编码错误：尝试使用 UTF-8，失败时回退到 latin-1，或使用
errors='replace'
常见问题解答
「文件格式处理：CSV、JSON 与 TXT」课时是免费的吗？
是的 — 「文件格式处理：CSV、JSON 与 TXT」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。
「文件格式处理：CSV、JSON 与 TXT」这节课中我会学到什么？
csv 模块、json.load/dump，以及适用于代理工具的安全文本编码。 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Agents 需要有经验吗？
无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 3 节课，共 4 节。
「文件格式处理：CSV、JSON 与 TXT」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 AI Agents 课中编写并运行代码吗？
能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。
此课程中的所有课时
- 在代理环境中读写文件
- 目录遍历与文件发现
- 文件格式处理：CSV、JSON 与 TXT
- 带错误处理的安全文件操作