Чтение и запись файлов в контексте агента

Почему операции с файлами важны для агентов

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)

Чтение файлов с помощью функции открытия

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)

Запись файлов с помощью функции открытия

# 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 — современные операции с файлами

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-файлов

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 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)

Работа с двоичными файлами

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)

Обработка проблем с кодировкой и символами новой строки

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))

Эффективная построчная обработка

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')

Создание каталогов и управление ими

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')

Рекомендации по работе агентов с путями к файлам

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)

Быстрая проверка: менеджеры контекста

Повторение операций ввода-вывода с файлами