Quy trình làm việc năng động

DynamicWorkflow dành cho trường hợp sự phối hợp giữa các agent phụ là công việc thực tế. Giả sử bạn có một vài chuyên gia -- một người đánh giá mã, một người tóm tắt các phát hiện, một người viết ghi chú cuối cùng. Mỗi cái đều dễ dàng gọi riêng; Phần khó nhất là cách thực hiện: xem lại ba tập tin cùng một lúc, chỉ giữ lại những báo cáo đã tìm thấy điều gì đó, tóm tắt chúng và giao bản tóm tắt cho người viết. Đạt được khả năng này khi việc điều phối đó bao gồm các vòng lặp phân ra, kết nối, biểu quyết hoặc thử lại mà bạn không muốn chạy lần lượt từng mô hình một, với mọi kết quả trung gian sẽ quay ngược lại qua ngữ cảnh của người điều phối.

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.

Ý tưởng

Cách thông thường để điều phối các subagent là một lệnh gọi công cụ cho mỗi bước. agent gọi người đánh giá và đợi, đọc kết quả, gọi lại người đánh giá, đợi lại, v.v. Mọi kết quả trung gian sẽ quay trở lại bối cảnh của agent và mỗi bước phụ thuộc vào kết quả trước đó là một lượt mô hình riêng biệt.

DynamicWorkflow đi một con đường khác. Bạn đưa cho nó một danh mục gồm các subagent được đặt tên và nó cung cấp cho mô hình một công cụ duy nhất, run_workflow. Bên trong công cụ đó, mô hình viết Python thông thường: mỗi subagent của bạn là một async chức năng mà nó có thể gọi, lặp lại và kết hợp. Tập lệnh sẽ chạy đến khi hoàn thành trong một lệnh gọi công cụ và chỉ giá trị cuối cùng của nó mới quay lại mô hình. Vũ đạo chuyển ra khỏi cuộc trò chuyện và chuyển sang mã.

Nếu bạn đã gặp Chế độ mã, điều này sẽ có cảm giác quen thuộc -- giống Monty sandbox và cùng một ý tưởng: viết một tập lệnh thay vì gọi nhiều công cụ. Sự khác biệt là những gì kịch bản được gọi. Trong Chế độ mã, nó gọi các công cụ riêng của agent; ở đây nó gọi toàn bộ các agent phụ.

Điều này liên quan đến Subagent như thế nào

harness có hai khả năng ủy quyền. Họ giao dịch bằng cùng một loại tiền -- được đặt tên, hoạt động của agent phụ biệt lập -- nhưng ở các cấp độ khác nhau:

• SubAgents để lộ một delegate_task(agent_name, task) công cụ. Mỗi đoàn có lệnh gọi công cụ riêng và lượt mô hình riêng. Nó phù hợp khi thỉnh thoảng có sự ủy quyền hoặc khi mỗi kết quả cần sự đánh giá của phụ huynh trước kết quả tiếp theo.
• DynamicWorkflow chuyển vũ đạo thành kịch bản. Tất cả các vòng lặp phân chia, kết nối, biểu quyết và thử lại đều chạy bên trong một lệnh gọi công cụ và các kết quả trung gian không bao giờ được đưa vào ngữ cảnh của công cụ gốc.

Bắt đầu với SubAgents nếu bạn không chắc chắn. A delegate_task người điều phối chuyển đổi thành một danh mục quy trình làm việc mà không tự thay đổi các subagent.

Cài đặt

Tập lệnh chạy bên trong hộp cát Monty, vì vậy hãy cài đặt phần bổ sung:

uv add "pydantic-ai-harness[dynamic-workflow]"

Quy trình công việc đầu tiên của bạn

Hai agent phụ, một người điều phối:

from pydantic_ai import Agent
from pydantic_ai_harness import DynamicWorkflow

reviewer = Agent('openai:gpt-5', name='reviewer', description='Reviews code for bugs.')
summarizer = Agent('openai:gpt-5', name='summarizer', description='Summarizes findings.')

orchestrator = Agent(
    'openai:gpt-5',
    capabilities=[DynamicWorkflow(agents=[reviewer, summarizer])],
)

reviewer và summarizer là những agent đơn giản -- giống nhau Agent bạn đã biết rồi. của họ name trở thành tên hàm mà mô hình gọi trong tập lệnh, vì vậy hãy chọn tên là mã định danh Python hợp lệ. của họ description cho model biết mỗi cái dùng để làm gì; viết nó theo cách bạn sẽ ghi lại một chức năng. DynamicWorkflow(agents=[...]) gộp chúng lại thành một khả năng và giao cho người điều phối một run_workflow công cụ.

Mô hình làm gì với nó

Khi người điều phối quyết định sử dụng công cụ này, nó sẽ không gọi từng agent phụ của bạn. Nó viết một kịch bản:

import asyncio

reports = await asyncio.gather(
    reviewer(task="Review auth.py for bugs:\n<file contents>"),
    reviewer(task="Review parser.py for bugs:\n<file contents>"),
)
await summarizer(task="Summarize these review findings:\n" + "\n\n".join(reports))

Các phần quan trọng:

• Mỗi agent phụ là một async chức năng. Bạn gọi nó với await.
• Bạn chuyển tác phẩm dưới dạng một đối số từ khóa duy nhất, task. Luôn theo từ khóa -- reviewer(task="..."), không reviewer("...").
• asyncio.gather(...) chạy hai đánh giá đồng thời thay vì lần lượt.
• Giá trị của biểu thức cuối cùng trở thành kết quả mà mô hình nhìn thấy. Trung gian reports danh sách không bao giờ rời khỏi hộp cát.

Mỗi cuộc gọi là một cuộc gọi đầy đủ Agent.run, với vòng lặp mô hình, lịch sử tin nhắn, công cụ và đầu ra được nhập riêng. Hai điều sau đây: các cuộc gọi được cô lập (một tổng đài viên phụ không nhớ gì về cuộc gọi trước đó, vì vậy hãy nhập mọi thứ họ cần vào task) và gọi chi phí token và mất thời gian (đó là lý do tại sao khả năng này cung cấp cho bạn ngân sách bên dưới).

agent phụ có thể trả về dữ liệu có cấu trúc

Một agent phụ trả lại bất cứ thứ gì output_type sản xuất. Mặc định là một chuỗi, nhưng cung cấp cho subagent một mô hình Pydantic và tập lệnh sẽ nhận được một dict:

from pydantic import BaseModel

class Score(BaseModel):
    value: int
    reason: str

critic = Agent('openai:gpt-5', name='critic', description='Scores an answer 0-10.', output_type=Score)

Bên trong tập lệnh, mô hình đọc các trường theo chỉ số dưới, giống như cách đọc đối tượng JSON:

result = await critic(task="Score this answer: ...")
result["value"]   # not result.value

Danh mục mà mô hình nhìn thấy hiển thị từng loại đầu ra dưới dạng TypedDict, vì vậy nó biết các trường và tự đọc chúng bằng chỉ số dưới.

Kết quả trở lại như thế nào

Giá trị của biểu thức cuối cùng của tập lệnh trở thành kết quả của công cụ -- mô hình không print() nó.

print() là để ghi nhật ký gỡ lỗi; nó xâu chuỗi lại, vì vậy hãy để biểu thức cuối cùng mang kết quả thực sự.

Lựa chọn mô hình agent phụ

Theo mặc định, mỗi subagent sử dụng mô hình mà nó được xây dựng. Đặt inherit_model=True khi máy chủ chuyển ghi đè mô hình mỗi lần chạy cho agent chính (ví dụ từ một /model command) và mọi công văn của subagent phải tuân theo mô hình gốc đã được giải quyết đó. Để nó đi False khi một subagent được cố tình ghim vào một mô hình khác.

Giữ an toàn: ngân sách

agent phụ không mang tính xác định, tính phí token và có thể phân bổ thành nhiều agent phụ hơn. DynamicWorkflow cung cấp cho bạn mức trần số lượng cứng, ngân sách token và bảo vệ chống lại các tập lệnh hộp cát chạy trốn.

max_agent_calls -- một con số chính xác

DynamicWorkflow(agents=[...], max_agent_calls=50)  # 50 is the default

Một mức trần cứng nhắc do máy chủ thực thi đối với số lần chạy subagent trong một lần chạy chính. Đó là một ngân sách được chia sẻ trên mọi run_workflow gọi trong lần chạy đó và nó vẫn giữ chính xác ngay cả khi tập lệnh kết thúc asyncio.gather. Khi hết ngân sách, quy trình làm việc sẽ ngừng gọi các agent phụ và trả về kết quả cuối cùng với các bản xem trước có giới hạn của tối đa 20 kết quả đã hoàn thành gần đây nhất. Đây là núm duy nhất giới hạn chính xác số lần chạy.

sub_agent_usage_limits và forward_usage - chi phí giới hạn

sub_agent_usage_limits là một UsageLimits áp dụng cho mỗi lần chạy agent phụ. forward_usage kiểm soát xem toàn bộ cây có chia sẻ một bộ đếm sử dụng hay không:

!!! cảnh báo “Cha mẹ run() giới hạn sử dụng không được chuyển tiếp" các usage_limits bạn chuyển cho phụ huynh run() không được chuyển tiếp tới các subagent -- nó chỉ được kiểm tra lại ở ranh giới yêu cầu của chính agent mẹ. Để ràng buộc các agent phụ, hãy đặt sub_agent_usage_limits; để có mức trần chính xác về số lần chạy, hãy sử dụng max_agent_calls.

resource_limits -- bảo vệ chính kịch bản

Các giới hạn này bảo vệ bộ nhớ của chính tập lệnh điều phối chứ không phải subagent mà nó gọi. Backstop mặc định là 256 MB không giới hạn thời gian. Đầu ra in được thu thập riêng với giới hạn mặc định 10 MiB của Monty.

DynamicWorkflow(agents=[...], resource_limits={'max_duration_secs': 30})

max_duration_secs đo thời gian tập lệnh của bạn dành để chạy mã hộp cát chứ không phải thời gian trên đồng hồ treo tường. Trong khi tập lệnh chờ trên một subagent thì nó bị tạm dừng và thời gian đó không được tính, do đó, giới hạn sẽ không kích hoạt quy trình làm việc bình thường cho dù subagent mất bao lâu. Công việc duy nhất của nó là bắt lỗi chạy trốn của CPU -- một while True: vòng lặp không bao giờ chờ đợi, điều mà không ngân sách agent phụ nào có thể dừng lại vì nó không bao giờ gọi agent phụ. Vượt qua 'unlimited' để loại bỏ mọi giới hạn tài nguyên hộp cát; nắp in 10 MiB riêng biệt vẫn còn. Một phần chính tả sẽ hợp nhất vào điểm dừng để bạn chỉ ghi đè các giới hạn mà bạn đặt tên.

Quy trình làm việc không lồng nhau

agent phụ không thể bắt đầu quy trình làm việc của riêng mình; một lồng nhau run_workflow cuộc gọi trả về lỗi đầu cuối thay vì chạy. Nguyên tắc thực tế: không cung cấp cho các agent phụ trong danh mục của bạn thông tin DynamicWorkflow khả năng. Họ là những người điều khiển dàn nhạc, không phải người điều phối.

Đổi tên agent phụ: WorkflowAgent

Theo mặc định, một agent phụ xuất hiện dưới quyền của chính nó name và description. Để đặt tên hoặc mô tả khác cho một quy trình công việc mà không cần chỉnh sửa agent, hãy gói nó trong một WorkflowAgent:

from pydantic_ai_harness.dynamic_workflow import WorkflowAgent

DynamicWorkflow(
    agents=[
        WorkflowAgent(
            reviewer,
            name='check',
            description='Checks one code change and returns actionable review findings.',
        ),
    ],
)

Bây giờ mô hình gọi check(task=...). Việc chuyển một agent trần là cách viết tắt của việc gói nó trong một WorkflowAgent không có ghi đè.

Thêm agent phụ vào giữa quá trình chạy: reveal()

Danh mục được cố định khi quá trình chạy bắt đầu, điều này sẽ giữ nó ở tiền tố bộ nhớ đệm nhắc nhở qua các lượt. Để cung cấp một subagent mới trong quá trình chạy (giả sử khi một agent sửa lỗi đã được cung cấp), hãy tham chiếu đến DynamicWorkflow dụ và gọi reveal():

workflow = DynamicWorkflow(agents=[reviewer])
orchestrator = Agent('openai:gpt-5', deps_type=MyDeps, capabilities=[workflow])

# later, from the host or from another tool:
workflow.reveal(fixer)

subagent được tiết lộ sẽ có thể gọi được ở bước tiếp theo; mô hình tìm hiểu về nó thông qua một tin nhắn thông báo ngắn mang chữ ký của chức năng mới. các run_workflow Bản thân mô tả vẫn bị đóng băng tại các agent có mặt khi quá trình chạy bắt đầu, do đó, tiết lộ runtime không bao giờ di chuyển tiền tố bộ đệm nhắc nhở. reveal() chỉ gắn thêm và xác thực ngay lập tức -- tên bị thiếu, số nhận dạng không hợp lệ, từ khóa dành riêng hoặc xung đột tên sẽ tăng lên UserError tại địa điểm cuộc gọi.

Chỉ tải nó khi cần: defer_loading

DynamicWorkflow mang theo một lượng văn bản hướng dẫn khá lớn và hầu hết các lượt chơi đều không cần đến nó. Giữ nó được thu gọn thành một mục nhập một dòng cho đến khi mô hình thực sự tải nó:

DynamicWorkflow(
    agents=[reviewer, summarizer],
    id='workflow',
    defer_loading=True,
)

defer_loading=True cần một nơi ổn định id. Xem khả năng theo yêu cầu để có hình ảnh đầy đủ.

Những gì chạy trong hộp cát

Tập lệnh chạy bằng Monty, một tập hợp con của Python. Biết các cạnh quan trọng:

• Không có thư viện của bên thứ ba.
• Các mô-đun thư viện tiêu chuẩn có thể nhập khẩu bao gồm sys, typing, asyncio, math, json, re, unicodedata, datetime, osvà pathlib. Nhập những gì bạn sử dụng. Các hoạt động của hệ thống tệp, môi trường và đồng hồ không được cấu hình cho tập lệnh quy trình công việc.
• Không có đồng hồ treo tường hoặc tính năng căn giờ -- không asyncio.sleep, không datetime.datetime.now(), không datetime.date.today(), và không time mô-đun.
• asyncio.gather(...) chạy các agent phụ đồng thời với các sản phẩm chờ đợi theo vị trí nhưng không có từ khóa lập luận, trong đó có return_exceptions=True. Các API tạo tác vụ và chờ đợi khác không khả dụng.

Trước khi chạy tập lệnh, nó được kiểm tra kiểu tĩnh đối với chữ ký của subagent. Một lỗi thông thường, có thể chứng minh được bằng phương pháp tĩnh, chẳng hạn như hàm sai chính tả, lỗi vị trí task, hoặc một đối số bị nhập sai sẽ tốn một lần thử lại nhưng không có ngân sách cho agent phụ hoặc thực thi hộp cát. Các giá trị được nhập dưới dạng Any có thể đạt được xác nhận runtime; họ vẫn bị từ chối trước khi một agent phụ chạy.

!!! cảnh báo "Lỗi chưa được phát hiện sẽ hủy bỏ toàn bộ tập lệnh" Sự cố của subagent xuất hiện dưới dạng RuntimeError. Kịch bản có thể bắt nó với try/except RuntimeError; một lỗi chưa được phát hiện sẽ hủy bỏ toàn bộ tập lệnh và mô hình sẽ thử lại tập lệnh đó. Nếu tập lệnh không thành công sau khi một số subagent đã hoàn tất, lời nhắc thử lại sẽ liệt kê các bản xem trước có giới hạn của tối đa 20 kết quả đã hoàn thành gần đây nhất. Mô hình có thể sử dụng lại bản xem trước không bị cắt bớt dưới dạng giá trị đơn giản thay vì phải trả tiền cho cùng một cuộc gọi.

Khả năng quan sát

Lửa rừng theo dõi là cách tốt nhất để xem quy trình làm việc đã làm gì. Mỗi lần chạy subagent xuất hiện được lồng trong run_workflow span và span mang chính xác code đối số mà mô hình đã viết để bạn có thể đọc đoạn mã mà nó thực sự chạy. Cho đến khi phát trực tuyến tiến độ hạng nhất, hãy thiết lập event_stream_handler trên mỗi agent phụ Agent để xem subagent chạy bên trong lệnh gọi một công cụ.

API

DynamicWorkflow(                  # all parameters are keyword-only
    agents=[...],                 # Sequence[AbstractAgent | WorkflowAgent], required
    tool_name='run_workflow',
    max_agent_calls=50,
    max_retries=3,
    forward_usage=True,
    inherit_model=False,          # True -> sub-agents run with the parent run's resolved model
    sub_agent_usage_limits=None,  # UsageLimits per sub-agent run; None -> pydantic-ai default
    resource_limits=None,         # None -> backstop (256 MB, no time cap);
                                  # 'unlimited' -> sandbox limits off; a dict merges onto the backstop
    id=None,                      # required when defer_loading=True
    description=None,             # one-line catalog entry shown while deferred
    defer_loading=False,
)

workflow.reveal(agent)            # AbstractAgent | WorkflowAgent; validates before appending

WorkflowAgent(
    agent,                        # AbstractAgent, required, positional
    name=None,                    # sandbox function name; falls back to agent.name
    description=None,             # function docstring; falls back to agent.description
)

DynamicWorkflowToolset và WorkflowResourceLimits cũng được xuất từ mô-đun để sử dụng nâng cao.

Nguồn: pydantic_ai_harness/dynamic_workflow/.

Đọc thêm

• Chế độ mã -- cùng một hộp cát, gọi các công cụ riêng của tổng đài viên thay vì agent phụ.
• agent phụ -- một subagent được ủy quyền cho mỗi lệnh gọi công cụ, không có kịch bản.
• Viết lại Bun bằng Rust (Bun) -- cùng một mẫu ở quy mô lớn, thông qua quy trình làm việc năng động của Claude Code.
• Khả năng và khả năng theo yêu cầu.