Bước kiên trì

StepPersistence ghi lại những gì một nhân viên đã làm ở mỗi ranh giới, tách biệt với việc liệu hoạt động có thể được tiếp tục một cách an toàn hay không. Nó là nền tảng bền bỉ cho những người điều phối ủy quyền cho các subagent -- ví dụ: người điều phối AICA tạo ra một code_librarian để điều tra một biểu tượng, sau đó tiếp tục cuộc điều tra của đại biểu đó bằng một câu hỏi tiếp theo.

Nó không phải là một điểm kiểm tra trạng thái biểu đồ đầy đủ. Khôi phục trạng thái khả năng, ảnh chụp nhanh không gian làm việc và tiếp tục nút biểu đồ nằm ngoài phạm vi và được theo dõi riêng (xem pydantic-ai-harness vấn đề #149 và #196).

Nguồn

Mặc dù Pydantic AI Harness có trên các bản phát hành 0.x nhưng API có thể thay đổi giữa các bản phát hành nhỏ; khi đó, cảnh báo ngừng sử dụng và hướng dẫn di chuyển ghi chú phát hành sẽ cho bạn (hoặc người đại diện của bạn) biết chính xác cách nâng cấp. Xem chính sách phiên bản.

Những gì nó mang lại cho bạn

1. Các sự kiện bước chỉ nối thêm. Mọi ranh giới thú vị (bắt đầu/kết thúc chạy, yêu cầu mô hình, lệnh gọi công cụ, lỗi) sẽ gắn thêm một StepEvent. Một cuộc gọi bị hủy giữa cuộc gọi công cụ vẫn để lại dấu vết sự kiện có thể sử dụng được.
2. Ảnh chụp nhanh liên tục. A ContinuableSnapshot được lưu tại các ranh giới nút đã xác định và một lần chạy không thành công sẽ lưu lịch sử trực tiếp của nó khi xảy ra lỗi. Mỗi ảnh chụp nhanh mang một state: complete khi mọi ToolCallPart có kết quả phù hợp, interrupted khi quá trình chụp giữ công cụ không ổn định (ví dụ: sự cố giữa chu kỳ công cụ). latest_snapshot và continue_run chỉ trả lại complete ảnh chụp nhanh trừ khi người gọi vượt qua include_interrupted=True. Vượt qua ảnh chụp nhanh messages quay lại Agent.run(message_history=...) để tiếp tục hoặc rẽ nhánh.
3. Sổ cái hiệu ứng công cụ. Vòng đời của mỗi lệnh gọi công cụ (started, completed, failed) được ghi lại theo (run_id, tool_call_id). Sau một sự cố, một công cụ có started bản ghi và không có bản cập nhật terminal nào được coi là unknown_after_crash: tác dụng phụ có thể xảy ra hoặc không.
4. Siêu dữ liệu dòng dõi. conversation_id (trình tự) và parent_run_id (phân cấp) là các trục độc lập. Xem Nhận dạng ba cấp độ.

Bắt đầu nhanh

import asyncio

from pydantic_ai import Agent
from pydantic_ai_harness import StepPersistence
from pydantic_ai_harness.step_persistence import InMemoryStepStore

store = InMemoryStepStore()
librarian = Agent(
    'openai:gpt-5',
    capabilities=[StepPersistence(store=store, agent_name='code_librarian')],
)


async def main():
    await librarian.run('Find ThinkingPartDelta and confirm the callable allowance')


asyncio.run(main())

Đó là toàn bộ thiết lập. run_id luôn luôn làAgent.run gọi, khớp với pydantic_ai RunContext.run_id. Để sử dụng nhóm logic nhiều lượt conversation_id= -- đó là nguyên hàm pydantic_ai-native của nó (xem Nhận dạng ba cấp độ).

run_id độ phân giải mỗi cuộc gọi:

• Rõ ràng run_id='libr-1' trở thành id cho một cuộc gọi này. Điều này phù hợp với các trường hợp sử dụng một lần (id xác định để kiểm tra, phát lại, gỡ lỗi hoặc chạy theo kịch bản một lần). Sử dụng lại một phiên bản khả năng có cùng nội dung rõ ràng run_id trên nhiều .run() cuộc gọi tăng ValueError trong before_run. Sổ cái hiệu ứng công cụ được khóa bởi (run_id, tool_call_id) và nhà cung cấp sử dụng lại id cuộc gọi công cụ xác định, do đó, một xung đột thầm lặng sẽ xóa unknown_after_crash tín hiệu. sử dụng conversation_id= thay vào đó là nhóm nhiều lượt.
• **agent_name thiết lập, run_id bỏ đặt ** xuất phát '{agent_name}-{8-char-hex}', mới được hiện thực hóa trong for_run mỗi .run() gọi. Việc sử dụng lại một phiên bản khả năng trong các lần chạy sẽ mang lại các id riêng biệt (code_librarian-a3b2, code_librarian-c9d1, v.v.). Đây là mặc định được đề xuất cho khả năng đại biểu.
• Không đặt quay lại ctx.run_id (id được tạo tự động của pydantic_ai) mỗi .run() cuộc gọi và tới UUID4 nếu cuộc gọi đó không có.

Mẫu điều phối -- một agent logic phục vụ nhiều lượt -- sử dụng conversation_id, không được chia sẻ run_id:

import asyncio

from pydantic_ai import Agent
from pydantic_ai_harness import StepPersistence
from pydantic_ai_harness.step_persistence import InMemoryStepStore

store = InMemoryStepStore()
orchestrator = Agent(
    'openai:gpt-5',
    capabilities=[StepPersistence(store=store, agent_name='orchestrator')],
)


async def main():
    for turn in turns:
        await orchestrator.run(turn, conversation_id='orch-conv')

    # All turns of this orchestrator, chronological:
    records = await store.list_runs(conversation_id='orch-conv')


asyncio.run(main())

Nhận dạng ba cấp độ

Khả năng phản ánh ngăn xếp danh tính của pydantic_ai:

StepEvent.conversation_id và RunRecord.conversation_id được cư trú từ ctx.conversation_id. Vì vậy ba .run() cuộc gọi chia sẻ một conversation_id sản xuất ba khác biệt run_ids, tất cả đều có thể truy vấn được dưới dạng một nhóm:

import asyncio


async def main():
    runs = await store.list_runs(conversation_id='conv-abc')  # 3 records, chronological


asyncio.run(main())

Tiếp tục điều tra đại biểu

pydantic_ai đã có rồi message_history= cho "tiếp tục với bối cảnh trước đó". StepPersistence không giới thiệu một cơ chế song song. Nó hiển thị một trình trợ giúp tải ảnh chụp nhanh đã giải quyết gần đây nhất:

import asyncio

from pydantic_ai import Agent
from pydantic_ai_harness import StepPersistence
from pydantic_ai_harness.step_persistence import InMemoryStepStore, continue_run

store = InMemoryStepStore()
librarian = Agent(
    'openai:gpt-5',
    capabilities=[StepPersistence(store=store, agent_name='code_librarian')],
)


async def main():
    # Earlier: tag the first turn with a conversation id so the follow-up can find it.
    await librarian.run(
        'Find ThinkingPartDelta and confirm the callable allowance',
        conversation_id='libr-conv',
    )

    # Later (possibly a different process):
    prior_run = (await store.list_runs(conversation_id='libr-conv'))[-1].run_id
    history = await continue_run(store, run_id=prior_run)
    await librarian.run(
        'Read _apply_provider_details_delta and check the path',
        message_history=history,
        conversation_id='libr-conv',   # keep the conversation grouping
    )


asyncio.run(main())

fork_run(store, run_id=...) trả về hình dạng tương tự nhưng được dự định khi người gọi muốn chạy logic phân nhánh từ điểm chụp nhanh đó (lần chạy mới có một phiên bản mới run_id và có lẽ là tươi conversation_id).

"An toàn để tiếp tục" nghĩa là gì

Theo mặc định continue_run trả về tin nhắn mới nhất complete ảnh chụp nhanh cho điều đó run_id -- một điểm mà công cụ của nó đã được giải quyết hoàn toàn khi được chụp. Ảnh chụp nhanh được viết ở các ranh giới này:

• sau mỗi lần CallToolsNode tất cả các cuộc gọi công cụ của họ đều được trả về -- yêu cầu trả lại công cụ đang chờ xử lý được gộp lại, do đó, điểm này sẽ ổn định ngay thời điểm công cụ hoàn thành, thậm chí trước khi yêu cầu mô hình tiếp theo được gửi đi,
• tại after_run, khi quá trình chạy kết thúc vượt qua ranh giới đó (một lần chạy không đạt đến ranh giới nào cả, hoặc một Agent.run_stream có phản hồi kết thúc sau câu trả lời cuối cùng) và
• khi một lần chạy thất bại: lịch sử trực tiếp tại thời điểm lỗi được lưu, bất kể hình dạng của nó -- một yêu cầu mô hình xuất hiện sau một chu trình công cụ sạch sẽ tạo ra một complete ảnh chụp nhanh; một sự cố xảy ra giữa chu kỳ công cụ tạo ra một interrupted một chiếc mang mọi chu trình đã hoàn thành.

Một interrupted ảnh chụp nhanh có thể được gửi trong sơ yếu lý lịch -- pydantic-ai (>= 2.10) sửa chữa việc ghép nối lệnh gọi công cụ/kết quả bị hỏng trước mỗi yêu cầu mô hình -- nhưng không nhất thiết an toàn: lệnh gọi công cụ đang chờ xử lý có thể được thực hiện lại (tiếp tục mà không có lời nhắc mới) hoặc kết thúc bằng lệnh gọi tổng hợp interrupted quay trở lại và không nói liệu tác dụng phụ ban đầu có xảy ra hay không. Đó là công việc của sổ cái hiệu ứng công cụ. Vì vậy, đường dẫn đọc mặc định bỏ qua interrupted ảnh chụp nhanh; vượt qua include_interrupted=True để continue_run / fork_run / latest_snapshot sau khi kiểm tra list_unresolved_tool_effects. Nếu không có ảnh chụp nhanh phù hợp, continue_run tăng lên LookupError.

Chạy dòng dõi: parent_run_id

parent_run_id là nhãn dòng dõi, không phải là phụ thuộc chức năng. Nó làm hai việc:

• Mỗi StepEvent và RunRecord mang nó, vì vậy bạn có thể lọc và nhóm.
• store.list_runs(parent_run_id='orch-1') trả về mọi lần chạy đại biểu trỏ vào người điều phối đó.

Nó được tự động suy ra cho việc ủy quyền trong quá trình: khi công cụ của người điều phối gọi đồng bộ quyền của đại biểu Agent.run(...), của đại biểu StepPersistence chọn người điều khiển dàn nhạc run_id thông qua một ContextVar đó là của người chỉ huy dàn nhạc wrap_run thiết lập. Không cần luồng:

import asyncio

from pydantic_ai import Agent
from pydantic_ai_harness import StepPersistence
from pydantic_ai_harness.step_persistence import InMemoryStepStore

store = InMemoryStepStore()
orchestrator = Agent(
    'openai:gpt-5',
    capabilities=[StepPersistence(store=store, agent_name='orchestrator')],
)
librarian = Agent(
    'openai:gpt-5',
    capabilities=[StepPersistence(store=store, agent_name='code_librarian')],
)


@orchestrator.tool_plain
async def ask_librarian(question: str) -> str:
    result = await librarian.run(question)   # parent_run_id auto-filled
    return result.output


async def main():
    # Tag the orchestrator turn so the lookup below can find its run_id.
    await orchestrator.run(
        'Where is ThinkingPartDelta defined?',
        conversation_id='orch-conv',
    )

    # All librarian runs now point at the orchestrator's run_id:
    orch_run_id = (await store.list_runs(conversation_id='orch-conv'))[-1].run_id
    delegates = await store.list_runs(parent_run_id=orch_run_id)


asyncio.run(main())

Đặt parent_run_id= rõ ràng để ghi đè (ví dụ: ủy quyền chéo quy trình trong đó ContextVars không lan truyền).

parent_run_id khác biệt với conversation_id. Người điều phối và đại biểu thường sống trong các cuộc trò chuyện khác nhau (người điều phối nói chuyện với người dùng; người đại diện nói chuyện với chính nó). Nhưng họ chia sẻ một liên kết cha-con.

Kiểm tra cây chạy

list_runs trả về các kết quả khớp được sắp xếp theo started_at tăng dần trên tất cả các chương trình phụ trợ -- chọn phiên bản gần đây nhất với [-1].

import asyncio


async def main():
    # Every delegate of one orchestrator run (chronological)
    delegates = await store.list_runs(parent_run_id='orch-3f2a')

    # Every run in one dialogue (multi-turn conversation across many .run() calls)
    turns = await store.list_runs(conversation_id='conv-abc')
    latest_turn = turns[-1]

    # Filters combine (AND):
    focused = await store.list_runs(
        parent_run_id='orch-3f2a',
        conversation_id='libr-conv',
    )

    # Detail per run:
    events = await store.list_events(run_id=delegates[0].run_id)
    snapshot = await store.latest_snapshot(run_id=delegates[0].run_id)
    unresolved = await store.list_unresolved_tool_effects(run_id=delegates[0].run_id)


asyncio.run(main())

Phục hồi lỗi

import asyncio


async def main():
    # An earlier delegate run died mid-investigation.
    events = await store.list_events(run_id='libr-3f2a')
    unresolved = await store.list_unresolved_tool_effects(run_id='libr-3f2a')
    for record in unresolved:
        # status == 'started' with no terminal update -- unknown_after_crash.
        print(f'tool {record.tool_name} ({record.tool_call_id}) may or may not have run')
        print(f'  idempotency_key={record.idempotency_key}  '
              f'effect_summary={record.effect_summary}')

    # Decide whether to resume or branch:
    history = await continue_run(store, run_id='libr-3f2a')
    # If the unresolved tools were read-only and safe to redo:
    await librarian.run('continue investigating', message_history=history,
                        conversation_id='libr-conv')
    # If side effects might have happened and the orchestrator wants a fresh attempt:
    history = await fork_run(store, run_id='libr-3f2a')
    # ... pass to a new delegate run with a different agent_name / conversation_id.

    # To resume from the interrupted frontier itself (the crashed cycle included),
    # after checking the unresolved effects above:
    history = await continue_run(store, run_id='libr-3f2a', include_interrupted=True)


asyncio.run(main())

Việc loại bỏ tác dụng phụ là trách nhiệm của người điều phối. Các công cụ ghi trạng thái bên ngoài phải chú thích quá trình hoạt động của chúng ToolEffectRecord qua annotate_tool_effect:

from pydantic_ai import RunContext
from pydantic_ai_harness.step_persistence import annotate_tool_effect


@orchestrator.tool
async def set_label(ctx: RunContext[Deps], issue: int, label: str) -> str:
    await annotate_tool_effect(
        store,
        ctx,
        idempotency_key=f'issue-{issue}::label::{label}',
        effect_summary=f'set label {label!r} on issue #{issue}',
    )
    await github.set_label(issue, label)   # the actual side effect
    return 'ok'

Người trợ giúp đọc hoạt động run_id từ StepPersistence ContextVar và tool_call_id / tool_name từ ctx, sau đó hợp nhất siêu dữ liệu vào bản ghi trước đó. Nó không hoạt động khi được gọi bên ngoài lệnh gọi công cụ được bao bọc theo từng bước. after_tool_execute bảo toàn cả hai trường khi nó ghi terminal completed / failed nhập cảnh.

Tay cầm biên nhận nén

StepPersistence.compaction_transcript_handle() phơi bày hiện tại run_id để thu gọn biên lai. Đây là mã định danh cho lịch sử chạy liên tục của cửa hàng này, không phải là lời hứa rằng bản ghi trước khi nén trước vẫn có sẵn: ảnh chụp nhanh có thể đã chứa lịch sử được nén và việc lưu giữ được định cấu hình có thể xóa các ảnh chụp nhanh cũ hơn.

Phụ trợ

• InMemoryStepStore -- quy trình cục bộ; tuyệt vời cho các bài kiểm tra.
• FileStepStore(directory) -- bố cục thư mục bên dưới <directory>/<run_id>/:
  • run.json -- RunRecord (dòng dõi)
  • events.jsonl -- chỉ nối thêm StepEvents
  • tool_effects.jsonl -- chỉ nối thêm ToolEffectRecords, nằm trong phạm vi lần chạy này
  • snapshots/{seq}.json -- ContinuableSnapshots, được đặt tên theo bộ đếm đơn điệu mỗi lần chạy (không phải step_index, sẽ va chạm khi giống nhau run_id được tái sử dụng trên Agent.run cuộc gọi, kể từ khi ctx.run_step đặt lại về 0 mỗi cuộc gọi).
• SqliteStepStore(database='runs.db') -- một tệp SQLite có các bảng runs, events, snapshots, tool_effects, và một anh chị em media bảng cho các đốm màu bên ngoài (xem Phương tiện lưu trữ bên dưới). Chế độ WAL được bật; tool_effects nâng cao mỗi (run_id, tool_call_id) vì vậy trạng thái mới nhất sẽ thắng; sử dụng ảnh chụp nhanh AUTOINCREMENT seq phản chiếu FileStepStore._next_snapshot_seq. Cơ sở dữ liệu được tạo trước khi chụp nhanh state cột đã tồn tại tự động lấy nó khi mở (các hàng hiện có được đọc là complete). Vượt qua connection= thay vì database= để chia sẻ một sqlite3.Connection với phần còn lại của đơn đăng ký của bạn; kết nối phải được mở bằng check_same_thread=False bởi vì các cuộc gọi hook được gửi đến một luồng công nhân.
• MongoStepStore(client= or db_url=, database=...) -- Bộ sưu tập MongoDB runs, events, snapshots, tool_effectsvà counters (nguyên tử $inc phân bổ sự đơn điệu seq). runs._id = run_id thực thi cú đánh duy nhất run_id hợp đồng. cần mongodb thêm (pip install pydantic-ai-harness[mongodb], cài đặt nào pymongo>=4.17.0); vượt qua một chia sẻ AsyncMongoClient như client=hoặc một chuỗi kết nối như db_url= (khi đó cửa hàng sẽ sở hữu khách hàng -- hãy gọi await store.aclose() để giải phóng nó). Các bộ phận riêng lẻ bằng hoặc cao hơn media_threshold_bytes ngoại hóa theo mặc định thành một MongoMediaStore trên cùng một khách hàng. Đó là mức giảm tải trên mỗi giá trị, không phải giới hạn tổng hợp: ảnh chụp nhanh của nhiều phần dưới ngưỡng vẫn có thể vượt quá giới hạn tài liệu 16 MiB của MongoDB và không thể chèn, vì vậy hãy giảm ngưỡng nếu điều đó gây rủi ro cho khối lượng công việc của bạn.

Tất cả thực hiện cùng một không đồng bộ StepStore giao thức, do đó, các hook khả năng không bao giờ chặn vòng lặp sự kiện trên phần phụ trợ tệp/sqlite (I/O được gửi qua anyio.to_thread); phần phụ trợ Mongo vốn không đồng bộ.

FileStepStore xác nhận run_id chống lại [A-Za-z0-9_.-]{1,200} (và từ chối ..) để ngăn chặn việc truyền tải đường đi. Người gọi chuyển ID do người dùng kiểm soát vẫn phải vệ sinh trước.

cái gì MongoStepStore tạo ra trong lần viết đầu tiên

Các vấn đề về cửa hàng createIndex trong lần viết đầu tiên, cho tám chỉ mục: conversation_id và parent_run_id (cả thưa thớt) cộng started_at trên runs; (run_id, seq) trên events; (run_id, seq) và (run_id, state, seq) trên snapshots; và độc đáo (run_id, tool_call_id) cộng thêm (run_id, status) trên tool_effects. Mặc định của nó MongoMediaStore thêm một cái nữa, được mô tả trên trang phương tiện. Ba hậu quả đáng lưu ý trước khi hướng cửa hàng vào quá trình triển khai hiện có:

• Người dùng kết nối cần có đặc quyền để tạo chỉ mục. Vai trò Atlas bị hạn chế không bị lỗi ở lần viết đầu tiên, không phải ở giai đoạn xây dựng.
• Việc xây dựng chỉ mục duy nhất không thành công nếu một tool_effects bộ sưu tập đã giữ bản sao (run_id, tool_call_id) cặp.
• Chỉ mục được xây dựng dựa trên thời gian chi phí và I/O của bộ sưu tập đã được điền sẵn trong lệnh gọi đầu tiên đó.

RunRecord.metadata và StepEvent.metadata được lưu trữ dưới dạng tài liệu lồng nhau, do đó khóa của chúng trở thành tên trường BSON: khóa chứa . hoặc bắt đầu bằng $ cần MongoDB 5.0 trở lên và khóa chứa byte NULL sẽ bị bộ mã hóa BSON từ chối trước khi nó đến máy chủ. CI thực hiện cả hai chương trình phụ trợ của Mongo chống lại mongo:8.

Tăng trưởng ảnh chụp nhanh giới hạn

Mỗi bước ghi một ảnh chụp nhanh toàn bộ lịch sử mới được khóa bằng số tăng dần seqvà không có gì được cắt bớt theo mặc định. Trong vòng một thời gian dài Agent.run số lượng ảnh chụp nhanh bằng với số bước gọi công cụ đã giải quyết, do đó, một lần chạy dài sẽ phải trả chi phí lưu trữ ngày càng tăng.

Cả bốn cửa hàng -- InMemoryStepStore, FileStepStore, SqliteStepStorevà MongoStepStore -- chấp nhận đăng ký tham gia max_snapshots_per_run: int | None (mặc định None, không giới hạn - tính theo từng byte của hành vi trước đó). Khi được đặt thành N >= 1, mỗi save_snapshot cắt bớt phần chạy xuống một bộ giữ lại:

• mới nhất N ảnh chụp nhanh bởi seq,
• ảnh chụp nhanh mới nhất về tổng thể (phục vụ latest_snapshot(include_interrupted=True)),
• mới nhất complete ảnh chụp nhanh (phục vụ đường dẫn đọc mặc định).

Hai chế độ cuối cùng giữ cho cả hai chế độ đọc đều chính xác ngay cả khi chế độ mới nhất N tất cả đều là ảnh chụp nhanh interrupted và bản lý lịch mới nhất complete nằm bên dưới cửa sổ đó, vì vậy bộ giữ lại có thể vượt quá N. from_spec(..., max_snapshots_per_run=N) chuyển tiếp ràng buộc tới cửa hàng mà nó xây dựng (backend='memory', 'file'hoặc 'sqlite'; cửa hàng Mongo được xây dựng trực tiếp chứ không phải từ thông số kỹ thuật).

from pydantic_ai_harness.step_persistence import FileStepStore

store = FileStepStore('runs', max_snapshots_per_run=8)

Việc cắt bớt một ảnh chụp nhanh không bao giờ xóa phương tiện bên ngoài của nó: các đốm màu có địa chỉ nội dung và có thể được chia sẻ giữa các ảnh chụp nhanh và các lần chạy, do đó GC mồ côi-blob nằm ngoài phạm vi (xem các mục tiêu không phải mục tiêu bên dưới). Thời hạn sử dụng dựa trên độ tuổi (TTL) cũng nằm ngoài phạm vi -- nó thuộc về mức độ chi tiết toàn bộ chứ không phải trên mỗi ảnh chụp nhanh.

Lưu giữ giới hạn sẽ loại bỏ các ảnh chụp nhanh theo từng bước cũ hơn, bao gồm cả các ảnh chụp nhanh được nén trước. Bất kỳ luồng xuôi nào tái tạo lại lịch sử bằng cách kết hợp các ảnh chụp nhanh được giữ lại của một lần chạy -- tìm kiếm ảnh chụp nhanh hoặc biên nhận nén được khóa run_id -- chỉ có thể nhìn thấy những gì được giữ lại. Với một giới hạn chặt chẽ (ví dụ max_snapshots_per_run=1) các trạng thái nén trước, cũ hơn đã không còn nữa, vì vậy, hãy coi giới hạn này là một giới hạn cứng về khoảng cách mà quá trình phục hồi đó có thể đạt tới. Để lại giới hạn tại Nonehoặc đặt nó đủ cao để bao gồm lịch sử bạn cần khôi phục khi việc tái thiết lịch sử trở nên quan trọng.

Phương tiện truyền thông bền bỉ

BinaryContent tải trọng (hình ảnh, âm thanh, tài liệu, video) được đặt dưới dạng base64 bên trong ảnh chụp nhanh sẽ làm nổi bật mọi tệp hoặc hàng chứa tin nhắn; một phần văn bản lớn (ví dụ: một chuỗi trả về công cụ lớn) cũng làm như vậy và có thể đẩy một MongoStepStore chụp nhanh qua giới hạn tài liệu 16 MiB của MongoDB (#440). Các tệp phụ trợ, sqlite và mongo sẽ loại bỏ bất kỳ BinaryContent.datavà bất kỳ phần nào có chuỗi content bằng hoặc cao hơn 64 KiB, thông qua cấu hình MediaStore, để lại tham chiếu URI trong ảnh chụp nhanh. giống nhau media_threshold_bytes quản lý nhị phân và văn bản như nhau; không có núm văn bản riêng biệt. Chuyến đi khứ hồi minh bạch: latest_snapshot(...).messages[*] trả lại bản gốc BinaryContent byte và văn bản.

Việc hiển thị bên ngoài văn bản không chỉ dành cho Mongo và không có lựa chọn không tham gia media_store=None: khung tập đi được dùng chung, do đó, một khung tập đi hiện có FileStepStore hoặc SqliteStepStore quá trình triển khai bắt đầu ghi các đốm màu cho các phần văn bản lớn cũng như các phần nhị phân kể từ bản phát hành này trở đi. Ảnh chụp nhanh được viết trước đó vẫn được khôi phục -- người đọc nhận ra hình dạng điểm đánh dấu nhị phân cũ hơn. Khả năng tương thích này chỉ dành cho bản nâng cấp: một bản phát hành có trước quá trình ngoại hóa văn bản coi mọi điểm đánh dấu là nhị phân, do đó, nó không thể xác thực ảnh chụp nhanh có chứa điểm đánh dấu văn bản bên ngoài. Giữ một trình đọc hiện tại để biết các ảnh chụp nhanh liên tục có chứa các điểm đánh dấu đó.

Ghi đè điểm đến bằng cách vượt qua điểm đến của riêng bạn MediaStore:

from pydantic_ai_harness.media import S3MediaStore
from pydantic_ai_harness.step_persistence import FileStepStore

store = FileStepStore(
    'runs',
    media_store=S3MediaStore(
        bucket='my-bucket',
        endpoint='https://<account>.r2.cloudflarestorage.com',
        region='auto',
        access_key_id=...,
        secret_access_key=...,
    ),
    media_threshold_bytes=64 * 1024,  # raise or lower if you want
)

Chọn không tham gia hoàn toàn (giữ byte nội tuyến trong ảnh chụp nhanh JSON/hàng):

from pydantic_ai_harness.step_persistence import FileStepStore, SqliteStepStore

FileStepStore('runs', media_store=None)
SqliteStepStore(database='runs.db', media_store=None)

URI là media+sha256://<hex>, có địa chỉ nội dung. Cùng một đốm màu được viết thông qua bất kỳ MediaStore giải quyết theo cách tương tự, do đó việc loại bỏ dữ liệu là tự động và việc di chuyển bộ nhớ cơ bản là trao đổi một dòng. Việc triển khai được vận chuyển là:

• DiskMediaStore(directory) -- một tập tin cho mỗi blob tại <directory>/<sha256>.bin.
• SqliteMediaStore(database=...) hoặc SqliteMediaStore(connection=...) -- một hàng trên mỗi đốm màu (INSERT OR IGNORE để khấu trừ địa chỉ nội dung).
• S3MediaStore(bucket=, endpoint=, region=, access_key_id=, secret_access_key=) -- URL kiểu đường dẫn cộng với SigV4 được quản lý bằng tay. Tương thích với AWS S3, Cloudflare R2 (region='auto'), MinIO và các nhà cung cấp tương thích với S3 khác. Chỉ PUT/GET/HEAD -- không có nhiều phần, vòng đời hoặc danh sách trong v1.
• MongoMediaStore(client= or db_url=, database=...) -- MongoDB, cần mongodb thêm. Mỗi đốm màu là các khối có địa chỉ sha256 trên một media tài liệu kê khai và anh chị em media_chunks bộ sưu tập (phân đoạn thủ công thay vì GridFS, do đó phần trích xuất được giữ nguyên -- xem trang phương tiện), do đó, một blob lớn hơn một tài liệu BSON vẫn lưu trữ và đọc lại. Phân đoạn giới hạn tài liệu, không phải bộ nhớ: không có API phát trực tuyến, vì vậy mỗi đốm màu được giữ toàn bộ trong bộ nhớ xử lý trên cả hai put và get. Bản kê khai giữ MediaContext.metadata nội tuyến và không bị phân đoạn, vì vậy hãy giữ siêu dữ liệu trên mỗi blob ở mức nhỏ. collection= đổi tên cả bộ sưu tập và chunk_size_bytes= (mặc định 8 MiB) đặt kích thước phân chia.

Hiển thị các byte bên ngoài dưới dạng URL

Mỗi cửa hàng chấp nhận một public_url= có thể gọi được, biến thành kinh điển media+sha256://<hex> URI thành URL mà mô hình có thể tìm nạp trực tiếp. sắp tới MediaExternalizer khả năng sẽ sử dụng điều này để trao đổi BinaryContent các bộ phận cho ImageUrl / AudioUrl / các phần URL khác trước khi mô hình nhìn thấy thông báo, cho phép nhà cung cấp tìm nạp phương tiện lớn qua mạng mà không cần mã hóa lại byte vào nội dung yêu cầu.

URL cơ sở tĩnh (nhóm R2 công khai, CDN):

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 chữ ký được chỉ định hoặc luân phiên -- chuyển bất kỳ lệnh gọi không đồng bộ nào cần (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, một túi có thể mở rộng cho mỗi lần phẫu thuật

Mỗi MediaStore phương pháp (put, get, exists, public_url, get_metadata) và cả các lệnh gọi do người dùng cung cấp (PublicUrlResolver, KeyStrategy) chấp nhận một 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

Tất cả các trường mặc định; các trường mới được thêm vào liên tục khi các trường hợp sử dụng xuất hiện. Vượt qua những gì bạn có, bỏ qua phần còn lại.

Sự kiên trì của cửa hàng. get_metadata(uri) chuyến đi khứ hồi do người dùng cung cấp metadata lập bản đồ trên tất cả bốn cửa hàng. media_type cũng được tồn tại nhưng không phải là một phần của những gì get_metadata trả về (nó được lưu trữ cho chính tải trọng byte, ví dụ như Content-Type).

• SqliteMediaStore viết metadata vào một cột JSON và media_type vào một cột chuyên dụng.
• S3MediaStore gửi metadata như đã ký x-amz-meta-* tiêu đề (tên khóa chữ và số ASCII cộng với dấu gạch ngang) và media_type như Content-Type; get_metadata đọc x-amz-meta-* giá trị trở lại từ phản hồi HEAD.
• DiskMediaStore ghi một tệp JSON sidecar (<resolved>.meta.json) cùng với mỗi đốm màu, nguyên tử thông qua tmp cộng với đổi tên. Sidecar chỉ vắng mặt khi vị trí không mang theo siêu dữ liệu.
• MongoMediaStore viết metadata dưới dạng một chuỗi JSON và media_type như một trường chuyên dụng trên tài liệu kê khai của blob ( media bộ sưu tập theo mặc định); get_metadata decode lại chuỗi JSON. Vì ánh xạ là một chuỗi JSON chứ không phải các trường lồng nhau nên các khóa siêu dữ liệu không tuân theo quy tắc tên trường BSON ở đây.

key_strategy: kiểm soát đường dẫn lưu trữ phụ trợ

Mặc định là <sha256>.bin. DiskMediaStore và S3MediaStore chấp nhận ghi đè để phù hợp với bố cục hiện có; SqliteMediaStore và MongoMediaStore thì không (thông báo là khóa chính của họ, do đó, khóa do người dùng chọn sẽ phá vỡ việc trích xuất hoặc không hoạt động -- hãy sử dụng table= / collection= để di chuyển các hàng hoặc tài liệu):

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)

Lưu ý: nếu chiến lược của bạn phụ thuộc vào context.media_type (ví dụ: để chọn tiện ích mở rộng), get(uri) và exists(uri) sẽ không tìm thấy blob trừ khi bối cảnh tương tự được cung cấp tại thời điểm đọc. Đối với các chiến lược tổ chức đường dẫn thuần túy (không phụ thuộc vào ngữ cảnh), ràng buộc này không được áp dụng.

DiskMediaStore từ chối các chiến lược tạo ra đường dẫn tuyệt đối hoặc đường dẫn chứa .. phân đoạn, để tránh thoát khỏi thư mục cửa hàng.

Riêng biệt, cả bốn cửa hàng đều chấp nhận một public_url= trình phân giải, hữu ích khi CDN, máy chủ HTTP cục bộ hoặc dịch vụ URL đã ký xếp trước các byte. Không có nó public_url(...) trả lại None (mô hình không bao giờ nhìn thấy URL trừ khi trình phân giải được định cấu hình và nó trả về một chuỗi).

Nhà cung cấp pydantic_ai tải xuống byte từ một URL một cách minh bạch khi mô hình đích không chấp nhận loại URL đó, do đó việc phát ra URL luôn an toàn: bạn chỉ mất khoản tiết kiệm dây, không bao giờ chính xác.

!!! lưu ý “Tương lai MediaExternalizer năng lực" Khi nó hạ cánh, thành phần sẽ là Agent(capabilities=[MediaExternalizer(store), StepPersistence(...)]) và StepPersistence sẽ thấy các thông báo đã được xác định bằng URL, vì vậy bước đi bên ngoài sẽ trở thành không hoạt động. API hiện tại không thay đổi.

Kiên trì với các chương trình phụ trợ không được hỗ trợ

DynamoDB, Postgres, Redis, GCS và các chương trình phụ trợ khác nằm ngoài phạm vi của bản phát hành này. Viết của riêng bạn StepStore (khoảng mười phương thức trên Giao thức) hoặc của riêng bạn MediaStore (năm phương pháp: put, get, exists, public_url, get_metadata) và chuyển nó qua store= / media_store=. Vui lòng mở một vấn đề nếu bạn gửi một vấn đề -- chúng tôi muốn cung cấp lớp bộ điều hợp dùng chung cuối cùng với N >= 3 triển khai thực tế trước khi trừu tượng hóa.

Khả năng này không làm được gì

• Nó không khôi phục khả năng ở trạng thái mỗi lần chạy, trạng thái nút biểu đồ, bộ đếm thử lại hoặc phản hồi phát trực tiếp trong chuyến bay.
• Nó không tự động loại bỏ các tác dụng phụ được phát lại. Các công cụ ghi các tạo phẩm, nhãn, PR hoặc trạng thái bên ngoài nên gọi annotate_tool_effect(store, ctx, ...) (xem Khôi phục lỗi) để người điều phối có thể quyết định xem việc phát lại có an toàn hay không.
• Nó không cắt bớt các sự kiện và theo mặc định không cắt bớt các ảnh chụp nhanh. Việc giữ chân là trách nhiệm của người gọi; tăng trưởng ảnh chụp nhanh có thể được giới hạn chọn tham gia với max_snapshots_per_run (xem Tăng trưởng ảnh chụp nhanh giới hạn).
• Nó không thu gom rác từ các phương tiện truyền thông bên ngoài. Việc cắt bớt một ảnh chụp nhanh sẽ giữ nguyên các đốm màu có địa chỉ nội dung của nó vì chúng có thể được chia sẻ giữa các ảnh chụp nhanh và các lần chạy.
• Nó không phát ra các nhịp OpenTelemetry. pydantic_ai's Instrumentation khả năng đã mở rộng agent run / chat / running tool và cư trú gen_ai.agent.name, gen_ai.agent.call.id, gen_ai.conversation.id qua hành lý. Một thay đổi trong tương lai có thể thêm các thuộc tính liên tục theo bước vào khoảng hoạt động; được theo dõi như một vấn đề tiếp theo.

Liên quan

• Tổng quan về khả năng
• Chế độ mã

Tham chiếu API

::: pydantic_ai_harness.step_persistence.StepPersistence