Implementation:Togethercomputer Together python Files Retrieve Content

Attribute	Value
Type	Implementation
Domains	Batch_Processing, Inference, API_Client
Repository	togethercomputer/together-python
Source	src/together/resources/files.py:L95-112
Last Updated	2026-02-15 16:00 GMT

Overview

Implementation of batch result file download via the Files.retrieve_content() method. Uses the DownloadManager for streaming download with progress tracking, returning a FileObject with the local file path and size.

Code Reference

def retrieve_content(
    self, id: str, *, output: Path | str | None = None
) -> FileObject:
    download_manager = DownloadManager(self._client)

    if isinstance(output, str):
        output = Path(output)

    downloaded_filename, file_size = download_manager.download(
        f"files/{id}/content", output, normalize_key(f"{id}.jsonl")
    )

    return FileObject(
        object="local",
        id=id,
        filename=downloaded_filename,
        size=file_size,
    )

Source: src/together/resources/files.py:L95-112

API Signature

Files.retrieve_content(id: str, *, output: Path | str | None = None) -> FileObject

Parameter	Type	Default	Description
`id`	`str`	(required)	The file identifier (e.g., `output_file_id` from a completed `BatchJob`)
`output`	str \| None	`None`	Optional local file path to save the downloaded content. If `None`, defaults to `{id}.jsonl`

I/O Contract

Input:

id -- A valid file identifier, typically the output_file_id or error_file_id from a completed BatchJob.
output (optional) -- A local path where the file should be saved.

Output: A FileObject with the following fields:

Field	Type	Description
`object`	`"local"`	Always set to `"local"` indicating a local file
`id`	string	The file identifier that was downloaded
`filename`	string	The local file path where the content was saved
`size`	int	The file size in bytes

Key Behavior

Default output filename: When no output path is specified, the file is saved using the pattern {id}.jsonl, where id is the file identifier passed through normalize_key() for safe filesystem naming:

downloaded_filename, file_size = download_manager.download(
    f"files/{id}/content", output, normalize_key(f"{id}.jsonl")
)

Streaming download: The method delegates to DownloadManager.download(), which performs a streaming HTTP download. This avoids loading the entire file into memory, which is important for large batch result files.

Path type coercion: If the output parameter is provided as a string, it is automatically converted to a Path object:

if isinstance(output, str):
    output = Path(output)

API endpoint: The download URL is constructed as files/{id}/content, which is a GET request to the Together AI file content endpoint.

Usage Examples

Downloading Batch Results to a Specific Path

from together import Together

client = Together()

# Assumes batch_job is a completed BatchJob object
result = client.files.retrieve_content(
    id=batch_job.output_file_id,
    output="results.jsonl",
)

print(f"Results saved to: {result.filename}")
print(f"File size: {result.size} bytes")

Downloading with Default Filename

from together import Together

client = Together()

# Downloads to {output_file_id}.jsonl in the current directory
result = client.files.retrieve_content(id="file-output-abc123")

print(f"Results saved to: {result.filename}")

Downloading Error File

from together import Together

client = Together()

batch_job = client.batches.get_batch("batch-abc123")

if batch_job.error_file_id:
    error_result = client.files.retrieve_content(
        id=batch_job.error_file_id,
        output="errors.jsonl",
    )
    print(f"Error file saved to: {error_result.filename}")

Complete End-to-End Batch Workflow

import time
import json
from together import Together

client = Together()

# Step 1: Upload the batch input file
file_response = client.files.upload(
    file="batch_input.jsonl",
    purpose="batch-api",
)

# Step 2: Create the batch job
batch_job = client.batches.create_batch(
    file_id=file_response.id,
    endpoint="/v1/chat/completions",
)

# Step 3: Poll for completion
while True:
    batch_job = client.batches.get_batch(batch_job.id)
    if batch_job.status == "COMPLETED":
        break
    elif batch_job.status in ("FAILED", "EXPIRED", "CANCELLED"):
        raise RuntimeError(f"Batch job {batch_job.status}: {batch_job.error}")
    time.sleep(60)

# Step 4: Download results
result = client.files.retrieve_content(
    id=batch_job.output_file_id,
    output="results.jsonl",
)

# Step 5: Parse results
with open(result.filename, "r") as f:
    for line in f:
        result_obj = json.loads(line)
        print(f"{result_obj['custom_id']}: {result_obj}")

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment