미디어 외부화

미디어 외부화 (Media Externalization)

Step Persistence가 실행 스냅샷을 작게 유지하는 데 쓰는 저장 배관이에요. 이것은 capability가 아니라 지원 인프라예요. 이 페이지의 어떤 것도 Agent(capabilities=[...])에 탑재되지 않아요. 큰 페이로드를 자동으로 외부화하는 스냅샷을 원한다면 Step Persistence부터 시작하세요. 스토어를 직접 구성하거나 워커 헬퍼를 직접 호출할 때 이 페이지를 읽으세요.

왜 존재하는가: 이미지·오디오·기타 BinaryContent를 나르는 대화는 그 바이트를 매 메시지에 인라인해요. 그리고 큰 텍스트 부분(큰 도구 반환 문자열 같은)도 똑같이 무겁죠. 그 히스토리를 영속하면 각 스냅샷이 페이로드를 다시 직렬화하고, 열 개 메시지가 참조하는 같은 이미지는 바이트의 열 개 복사본이에요. 콘텐츠 주소 지정(content-addressed) 스토어는 각 페이로드를 자기 해시로 키해 한 번 쓰고, 그 자리에 짧은 media+sha256:// URI를 남겨요.

출처: 문서

본문

임포트 경로

이 헬퍼들은 서브모듈에서 임포트하세요. 최상위 pydantic_ai_harness 재수출은 없어요:

from pydantic_ai_harness.media import (
    DiskMediaStore,
    S3MediaStore,
    SqliteMediaStore,
    externalize_media,
    restore_media,
)

Pydantic AI Harness가 0.x 릴리스인 동안 API는 마이너 릴리스 사이에 바뀔 수 있어요. 바뀔 때는 폐기 경고와 릴리스 노트 마이그레이션 안내가 정확히 어떻게 업그레이드할지 알려줘요. 버전 정책 참고.

소비자 (Consumers)

StepPersistence는 이 스토어들을 file·sqlite·mongo 백엔드로 실행 스냅샷의 큰 BinaryContent와 텍스트 부분을 외부화하는 데 사용해요 (미디어 영속 참고).

왜 콘텐츠 주소 지정인가 (Why content-addressing)

URI는 페이로드 해시에서 파생되므로, 동일한 바이트는 자동으로 중복 제거돼요. 같은 바이트는 얼마나 많은 메시지·스냅샷이 참조하든 한 번 저장되고, URI가 바뀌지 않으므로 기반 스토리지를 옮기는 것은 한 줄 교체예요.

스토어 (Stores)

모든 스토어는 MediaStore 프로토콜을 구현해요 — put, get, exists, public_url, get_metadata, 모두 비동기·콘텐츠 주소 지정.

스토어	기반	사용 시점
`DiskMediaStore(directory=...)`	디스크의 디렉터리	로컬 실행과 테스트
`SqliteMediaStore(database=...)`	SQLite 데이터베이스	데이터와 함께 여행하는 단일 파일 스토어
`S3MediaStore(bucket=, endpoint=, region=, ...)`	S3 또는 S3 호환 버킷	공유 또는 프로덕션 스토리지
`MongoMediaStore(client= or db_url=, database=, ...)`	MongoDB (sha256 주소의 수동 청킹)	MongoDB 배포; BSON 문서 하나보다 큰 blob

S3MediaStore는 path-style URL과 직접 만든 SigV4를 사용하므로 AWS S3, Cloudflare R2(region='auto'), MinIO, 다른 S3 호환 프로바이더와 호환돼요. SqliteMediaStore는 database= 대신 connection=도 받아 sqlite3.Connection을 공유할 수 있어요.

MongoMediaStore는 mongodb 엑스트라(pymongo>=4.17.0 설치)가 필요해요. 공유 AsyncMongoClient를 client=로, 또는 연결 문자열을 db_url=로 넘기세요(그러면 스토어가 클라이언트를 소유합니다 — 해제하려면 await store.aclose() 호출). database=는 항상 필요해요. 각 blob은 media_chunks 컬렉션에 sha256 주소의 청크로 저장되고, blob당 media 매니페스트 문서(_id = <digest>)가 있어요. 청킹이 각 BSON 문서를 경계 지어서, MongoDB의 16 MiB 문서 상한보다 큰 blob도 저장·읽기가 되돌아와요. 메모리는 경계 짓지 않아요. put은 전체 페이로드를 bytes로 받고 get은 모든 청크를 하나의 bytearray로 재조립하므로, blob은 양방향으로 프로세스 메모리에 맞아야 해요. 스트리밍 API는 없어요. 매니페스트는 MediaContext.metadata를 인라인으로 담고 청킹되지 않으므로, blob당 메타데이터를 작게 유지하세요. GridFS 드라이버 대신 수동 청킹을 의도적으로 써요. digest가 매니페스트 _id라 동일한 바이트가 중복 제거되고(GridFS는 ObjectId로 파일을 키하고 중복 제거를 안 함), 평문 컬렉션 표면이 메모리 안에서 완전히 테스트 가능하게 유지되기 때문이에요.

pip install "pydantic-ai-harness[mongodb]"

uv add "pydantic-ai-harness[mongodb]"

두 생성자 손잡이가 그 레이아웃을 형태화해요. collection=(기본 'media')은 매니페스트 컬렉션을 이름 짓고 청크 컬렉션을 <collection>_chunks로 파생하며, A-Za-z_* 밖의 이름은 거부돼요. chunk_size_bytes=(기본 8 MiB)는 분할 크기를 설정하고, 1바이트 미만 또는 16 MiB에서 청크 문서 자체 필드의 64 KiB 헤드룸을 뺀 것 이상은 거부돼요. 더 큰 청크는 삽입 시 MongoDB가 거부하는 문서를 만들기 때문이에요.

첫 put 또는 get에서 스토어는 청크 컬렉션에 복합 (files_id, n) 인덱스에 대한 createIndex를 발행해요. 그것 없이는 재조립이 컬렉션 스캔이 되거든요. 그러므로 연결 사용자는 인덱스 생성 권한이 필요해요. 제한된 Atlas 역할에는 없을 수 있고, 이미 채워진 컬렉션을 가리키면 그 첫 호출에서 인덱스 빌드 비용을 지불해요.

from pymongo import AsyncMongoClient

from pydantic_ai_harness.media import MongoMediaStore

client = AsyncMongoClient('mongodb://localhost:27017')
store = MongoMediaStore(client=client, database='agent_media')

워커 헬퍼 (Walker helpers)

externalize_media와 restore_media는 메시지 노드를 걸어가며 페이로드를 URI로·그리고 다시 바꿔요:

from pydantic_ai_harness.media import DiskMediaStore, externalize_media, restore_media

store = DiskMediaStore(directory='./media')

# Replace binary and text payloads at or above the threshold with media+sha256:// URIs.
lean = await externalize_media(message, media_store=store, threshold_bytes=32_000)

# Later, rehydrate the URIs back into the original parts.
full = await restore_media(lean, media_store=store)

externalize_media는 큰 BinaryContent와 큰 텍스트 둘 다 외부화해요. 그 문자열 content가 threshold_bytes UTF-8 바이트에 닿는 어떤 메시지 부분(TextPart, ThinkingPart, 문자열 반환 ToolReturnPart, 문자열 값 UserPromptPart)과, UserPromptPart.content 시퀀스나 ToolReturn 안에 들어 다니는 어떤 TextContent 요소까지요. 같은 threshold_bytes가 바이너리와 텍스트를 지배하고, 아래 페이로드는 인라인으로 유지돼요. 왕복은 투명해요. restore_media가 바이너리 바이트와 텍스트를 대칭으로 다시 인라인하죠. 미디어를 직접 키해야 한다면 media_uri_for와 parse_media_uri가 원시 URI 왕복을 줘요.

현재 리더는 텍스트 외부화 전에 쓰인 바이너리 마커를 복원해요. 그 호환성은 업그레이드 전용이에요. 텍스트 외부화보다 이전 릴리스는 모든 마커를 바이너리로 취급하므로, 외부화된 텍스트 마커를 담은 스냅샷을 검증할 수 없어요. 그 마커들을 담은 영속 스냅샷에는 현재 리더를 유지하세요.

페이로드가 마커 형식과 같은 네임스페이스 키를 쓰면, 기록자는 그 값을 버전화된 예약 매핑으로 옮기고 현재 리더가 복원해요. 이 이스케이핑 형식이 추가되기 전에 쓰인 마커도 여전히 복원돼요. 리더는 예약 매핑이 기록자가 만들 모양을 갖고 버전 키가 형식 자체 네임스페이스의 스탬프를 담을 때만 예약 키를 자체 것으로 취급하므로, 우연히 어느 하나 또는 둘을 담은 페이로드는 그대로 둬요. 둘 다 담고 이 리더가 모르는 버전으로 스탬프된 마커는, 예약 값을 제거한 채 복원되기보다 거부돼요. 다른 방향 호환성은 업그레이드 전용이에요. 이스케이핑 형식보다 이전 리더는 외부화된 필드를 올바르게 다시 인라인하지만, 호출자의 값을 자기 키로 복원하는 대신 예약 매핑에 남겨 두고 두 예약 키도 페이로드에 남겨요. 이스케이프된 마커들을 담은 영속 스냅샷에는 현재 리더를 유지하세요.

공개 URL (Public URLs)

스토어가 CDN·로컬 HTTP 서버·서명 URL 서비스 앞에 있을 때 public_url= 리졸버를 넘기거나(make_static_public_url 사용) 저장된 media+sha256:// URI를 모델이 직접 fetch할 수 있는 URL로 바꿔요. 리졸버가 없으면 public_url(...)은 None을 반환해요.

공개 버킷·CDN용 정적 베이스 URL:

from pydantic_ai_harness.media import S3MediaStore, make_static_public_url

store = S3MediaStore(
    bucket='my-bucket',
    endpoint='https://<acc>.r2.cloudflarestorage.com',
    region='auto',
    access_key_id=..., secret_access_key=...,
    key_prefix='media/',
    public_url=make_static_public_url('https://pub-abc.r2.dev', key_prefix='media/'),
)

사전 서명되거나 회전 서명 URL — (uri, MediaContext)를 받는 어떤 비동기 호출 가능이든:

from pydantic_ai_harness.media import MediaContext, S3MediaStore


async def presign(uri: str, ctx: MediaContext) -> str:
    key = 'media/' + uri.removeprefix('media+sha256://') + '.bin'
    return await my_signer.generate(key, ttl=3600, content_type=ctx.media_type)


store = S3MediaStore(..., public_url=presign)

`MediaContext`

모든 스토어 메서드와 두 사용자 제공 호출 가능(PublicUrlResolver, KeyStrategy)은 확장 가능한 연산별 가방인 MediaContext를 받아요:

from collections.abc import Mapping
from dataclasses import dataclass, field


@dataclass(frozen=True, kw_only=True)
class MediaContext:
    media_type: str | None = None                    # e.g. 'image/png'
    filename: str | None = None                      # original filename, when known
    metadata: Mapping[str, str] = field(default_factory=dict)  # user-supplied tags

모든 필드가 기본값이라, 가진 것을 넘기고 나머지는 무시해요. 새 필드는 사용 사례가 생기면 논브레이킹으로 추가돼요. get_metadata(uri)은 네 스토어 모두에서 사용자 제공 metadata 매핑을 왕복시키고, media_type은 별도로 영속돼요(바이트 페이로드의 Content-Type으로).

`KeyStrategy`

온-스토어 키 레이아웃 기본값은 <sha256>.bin이에요. DiskMediaStore와 S3MediaStore는 기존 레이아웃에 맞추는 key_strategy= 오버라이드를 받아요. SqliteMediaStore와 MongoMediaStore는 digest가 그들의 기본 키라 받지 않아요. 행·문서를 옮기려면 table=/collection=을 쓰세요:

from pydantic_ai_harness.media import DiskMediaStore, MediaContext


def by_media_type(uri: str, ctx: MediaContext) -> str:
    digest = uri.removeprefix('media+sha256://')
    ext = {'image/png': '.png', 'image/jpeg': '.jpg'}.get(ctx.media_type or '', '.bin')
    return f'images/{digest}{ext}'


store = DiskMediaStore('runs', key_strategy=by_media_type)

전략이 ctx.media_type에 의존하면, get/exists가 blob을 찾도록 읽을 때 같은 컨텍스트를 공급해야 해요. DiskMediaStore는 절대 경로나 .. 세그먼트를 만드는 전략을 거부해, 쓰기를 스토어 디렉터리 안에 유지해요. 그 위에 쌓고 싶으면 default_key_strategy가 내보내져요.

API

심볼	용도
`MediaStore`	비동기 콘텐츠 주소 스토어 프로토콜 (`put` / `get` / `exists` / `public_url` / `get_metadata`)
`DiskMediaStore`, `SqliteMediaStore`, `S3MediaStore`, `MongoMediaStore`	구체 스토어(`MongoMediaStore`는 `mongodb` 엑스트라 필요)
`MediaContext`	스토어 연산을 관통하는 연산별 컨텍스트(미디어 타입, 파일명, 태그)
`KeyStrategy`, `default_key_strategy`	온-스토어 키 레이아웃
`PublicUrlResolver`, `make_static_public_url`	저장된 URI를 공개 URL로 해석
`externalize_media`, `restore_media`	메시지 노드를 걸어 큰 바이너리·텍스트 페이로드 외부화/재수화
`media_uri_for`, `parse_media_uri`	`media+sha256://` URI 계산·파싱

소스: pydantic_ai_harness/media/.

Step Persistence — 이 스토어들의 첫 소비자. 실행 스냅샷에서 큰 BinaryContent·텍스트 부분을 외부화.

더 알아보기 (Learn more)

Step Persistence — 이 스토어들의 소비자.
Pydantic AI Harness — 패키지 전반.

미디어 외부화