Mengontrol ketersediaan alat

Halaman ini mencakup tiga teknik pelengkap untuk mengontrol alat mana yang dapat dipanggil model dan dalam urutan apa, semuanya dalam satu agen berjalan, tanpa memerlukan alur kerja:

• Paparan alat progresif — menambahkan atau menghapus alat saat runtime dari dalam alat atau middleware fungsi, sehingga model hanya melihat alat yang siap digunakan.
• Middleware gating — gunakan middleware fungsi untuk memvalidasi argumen panggilan dan mengembalikan umpan balik korektif tanpa menjalankan fungsi yang mendasar.
• Panggilan pertama paksa — gunakan tool_choice untuk mengharuskan model memanggil alat tertentu sebelum yang lain.

Penampilan alat secara bertahap

Eksposur alat progresif memungkinkan Anda memulai eksekusi dengan sekumpulan alat kecil dan menambahkan atau menghapus alat sebagai respons terhadap hasil alat sebelumnya, semuanya dalam proses yang sama. Model hanya akan melihat kumpulan yang diperbarui pada iterasi berikutnya dari loop pemanggilan fungsi; pemanggilan tool yang sudah diminta dalam batch yang sedang diproses tetap akan dieksekusi sebelum perubahan tersebut berlaku.

API bersifat eksperimental dan hidup di FunctionInvocationContext:

Kedua fungsi pembantu menghasilkan ExperimentalWarning saat pertama kali dipanggil dalam suatu proses (ID fitur PROGRESSIVE_TOOLS). Memanggil salah satu fungsi pembantu di luar loop pemanggilan fungsi akan menghasilkan RuntimeError.

Pola alat pemuat

Daftarkan sekumpulan kecil alat "loader" sejak awal dan biarkan model memuat alat tambahan sesuai permintaan. Ini menjaga skema awal tetap kecil, yang meningkatkan akurasi pemilihan alat dan mengurangi biaya.

import asyncio
import warnings
from typing import Annotated

from agent_framework import Agent, FunctionInvocationContext, tool
from agent_framework.openai import OpenAIChatClient
from pydantic import Field

warnings.filterwarnings("ignore", category=FutureWarning)  # suppress ExperimentalWarning for brevity


@tool(approval_mode="never_require")
def factorial(n: Annotated[int, Field(description="A non-negative integer.")]) -> str:
    """Compute the factorial of n."""
    if n < 0:
        return "Error: n must be a non-negative integer."
    result = 1
    for value in range(2, n + 1):
        result *= value
    return f"{n}! = {result}"


@tool(approval_mode="never_require")
def fibonacci(n: Annotated[int, Field(description="The 0-based index in the Fibonacci sequence.")]) -> str:
    """Compute the n-th Fibonacci number."""
    if n < 0:
        return "Error: n must be a non-negative integer."
    a, b = 0, 1
    for _ in range(n):
        a, b = b, a + b
    return f"fib({n}) = {a}"


# The ctx parameter is injected by the framework and is NOT visible to the model.
@tool(approval_mode="never_require")
def load_math_tools(ctx: FunctionInvocationContext) -> str:
    """Load additional math tools (factorial, fibonacci) so they can be used."""
    ctx.add_tools([factorial, fibonacci])
    return "Loaded math tools: factorial, fibonacci. You can now call them."


async def main() -> None:
    agent = Agent(
        client=OpenAIChatClient(),
        name="MathAgent",
        instructions=(
            "You are a math assistant. "
            "If you need math capabilities that are not yet available, call load_math_tools first."
        ),
        tools=[load_math_tools],  # agent starts with only the loader
    )
    print(await agent.run("What is 5 factorial?"))


asyncio.run(main())

Sampel lengkap yang dapat dijalankan adalah di python/samples/02-agents/tools/dynamic_tool_exposure.py.

Pola penyaringan

Daftarkan hanya alat pembaca terlebih dahulu. Tool baca menambahkan tool tulis setelah pengambilan data berhasil, sehingga model tidak dapat memanggil tool tulis sebelum tool baca dijalankan.

from agent_framework import Agent, FunctionInvocationContext, tool
from agent_framework.openai import OpenAIChatClient

_last_fetched_id: str | None = None


@tool(approval_mode="never_require")
def get_record(record_id: str, ctx: FunctionInvocationContext) -> str:
    """Fetch a record. Unlocks update_record for the same record."""
    global _last_fetched_id
    _last_fetched_id = record_id
    ctx.add_tools(update_record)  # gate: expose the write tool now
    return f"Record {record_id}: title='Example record', status='open'"


@tool(approval_mode="never_require")
def update_record(record_id: str, status: str) -> str:
    """Update the status of a record."""
    return f"Updated record {record_id} to status '{status}'."


agent = Agent(
    client=OpenAIChatClient(),
    name="RecordAgent",
    instructions="You help manage records. Fetch a record before updating it.",
    tools=[get_record],  # update_record is hidden until get_record runs
)

Karena ctx.tools diatur ulang ke [get_record] pada awal setiap kali dijalankan, gerbang diaktifkan kembali secara otomatis pada setiap giliran percakapan.

Pengendalian akses middleware

Middleware fungsi dapat memeriksa argumen panggilan alat yang masih tertunda dan menolaknya sebelum fungsi dasarnya dijalankan dengan menetapkan context.result tanpa memanggil call_next(). String yang ditetapkan untuk context.result dikembalikan ke model sebagai hasil fungsi, memberikan umpan balik korektif.

Ini berguna untuk pemeriksaan pada tingkat argumen yang memerlukan informasi yang tidak tersedia saat pendefinisian skema, misalnya memastikan bahwa pembaruan ditujukan ke item yang sama dengan yang diambil sebelumnya dalam proses eksekusi.

from collections.abc import Awaitable, Callable

from agent_framework import FunctionInvocationContext

_last_fetched_id: str | None = None


async def enforce_read_before_write(
    context: FunctionInvocationContext,
    call_next: Callable[[], Awaitable[None]],
) -> None:
    """Reject update_record calls that target a different record than the one fetched."""
    if context.function.name == "update_record":
        requested_id = context.arguments.get("record_id") if hasattr(context.arguments, "get") else None
        if requested_id != _last_fetched_id:
            # Set result without calling call_next — the function never executes.
            context.result = (
                f"Error: you must fetch record '{requested_id}' before updating it. "
                f"Last fetched record was '{_last_fetched_id}'."
            )
            return
    await call_next()

Tambahkan middleware ke agen:

agent = Agent(
    client=OpenAIChatClient(),
    name="RecordAgent",
    instructions="Fetch a record before updating it.",
    tools=[get_record, update_record],
    middleware=[enforce_read_before_write],
)

Untuk informasi lebih lanjut tentang middleware untuk fungsi, lihat Mendefinisikan Middleware dan Penggantian Hasil.

Memaksa pemanggilan alat dengan tool_choice

Untuk mengharuskan model memanggil alat tertentu sebagai tindakan pertamanya, teruskan tool_choice dengan mode "required" dan required_function_name. Kerangka kerja secara otomatis direset tool_choice ke None setelah iterasi pertama sehingga model bebas pada iterasi berikutnya.

result = await agent.run(
    "Update record REC-42 to status 'in-progress'.",
    options={"tool_choice": {"mode": "required", "required_function_name": "get_record"}},
)

Bidang tool_choice menerima ToolMode dict, atau bentuk singkat berupa string "auto", "required", atau "none":

from agent_framework import ToolMode

tool_choice: ToolMode = {"mode": "required", "required_function_name": "get_record"}

Semantik dan peringatan