Eszköz rendelkezésre állásának szabályozása

Ez az oldal három kiegészítő technikát tartalmaz annak szabályozására, hogy a modell mely eszközöket hívhatja meg, és milyen sorrendben, mindezt egyetlen ügynök futtatásával, munkafolyamat megkövetelése nélkül:

• Progresszív eszközexpozíció – eszközök hozzáadása vagy eltávolítása futtatókörnyezetben egy eszköz vagy függvény köztes szoftverén belülről, így a modell csak a használatra kész eszközöket látja.
• Köztes szoftver üzembe helyezése – függvényközvetítő használatával érvényesítheti a hívási argumentumokat, és korrekciós visszajelzést küldhet vissza az alapul szolgáló függvény végrehajtása nélkül.
• Első hívás kikényszerítése – a(z) tool_choice használatával megkövetelhető, hogy a modell egy adott eszközt bármely másik előtt hívjon meg.

Eszközök fokozatos megjelenítése

A progresszív eszközexpozíció lehetővé teszi a futtatás elindítását egy kis eszközkészlettel, és eszközöket adhat hozzá vagy távolíthat el a korábbi eszközeredmények alapján, mindezt ugyanazon a futtatáson belül. A modell csak a függvényhívási ciklus következő iterációján látja a frissített készletet; a repülés közbeni kötegben már kért eszközhívások a módosítás érvénybe lépése előtt is futnak.

Az API kísérleti jellegű, és a következőn FunctionInvocationContextél:

Mindkét segítő kibocsátja ExperimentalWarning az első alkalommal, amikor egy folyamat meghívja őket (funkcióazonosító PROGRESSIVE_TOOLS). Bármelyik segédfüggvény meghívása a függvényhívási cikluson kívül RuntimeError kivételt eredményez.

Betöltőeszköz-minta

Regisztráljon előre egy kisebb „betöltőeszköz”-készletet, és hagyja, hogy a modell igény szerint további eszközöket töltsön be. Így a kezdeti séma kicsi marad, ami javítja az eszközök kiválasztásának pontosságát és csökkenti a költségeket.

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())

A teljes futtatható minta a következő helyen python/samples/02-agents/tools/dynamic_tool_exposure.pytalálható: .

Kapuzási minta

Kezdetben csak az olvasóeszközt regisztrálja. Az olvasási eszköz egy sikeres beolvasás után hozzáadja az íróeszközt, így a modell nem tudja meghívni az íróeszközt az olvasóeszköz futtatása előtt.

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
)

Mivel a ctx.tools minden futtatás elején visszaáll a(z) [get_record] értékre, a kapu minden beszélgetési fordulóhoz automatikusan újraélesedik.

Köztes szoftver korlátozása

A függvényköztes réteg megvizsgálhatja egy függőben lévő eszközhívás argumentumait, és elutasíthatja azt, mielőtt az alapul szolgáló függvény végrehajtódna, úgy hogy beállítja a(z) context.result elemet anélkül, hogy meghívná a(z) call_next() elemet. A függvény eredményeként a rendszer visszaadja a modellhez hozzárendelt context.result sztringet, és korrekciós visszajelzést ad neki.

Ez olyan argumentumszintű ellenőrzésekhez hasznos, amelyekhez sémadefiníciós időpontban nem elérhető információkra van szükség, például annak ellenőrzéséhez, hogy egy frissítés ugyanazt az elemet célozza-e meg, amelyet a futtatás korábbi szakaszában lekért.

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()

Adja hozzá a köztes szoftvert az ügynökhöz:

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

A köztes szoftverekkel kapcsolatos további információkért lásd a Köztes szoftver és az Eredmény felülbírálásának definiálása című témakört.

Eszközhívás kényszerítése a következővel: tool_choice

Ha azt szeretné, hogy a modell első lépésként egy adott eszközt hívjon meg, adja át a(z) tool_choice paramétert "required" móddal és egy required_function_name értékkel. A keretrendszer az első iteráció után automatikusan a tool_choice értékét None értékre állítja vissza, így a modell a következő iterációkban szabaddá válik.

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

A tool_choice mező elfogad egy ToolMode szótárat, vagy a következő rövidített karakterláncokat: "auto", "required" vagy "none":

from agent_framework import ToolMode

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

Szemantikák és kikötések