|
|
from __future__ import annotations |
|
|
|
|
|
import inspect |
|
|
from dataclasses import dataclass |
|
|
from typing import TYPE_CHECKING, Any, Callable |
|
|
|
|
|
from openai.types.responses.response_prompt_param import ( |
|
|
ResponsePromptParam, |
|
|
Variables as ResponsesPromptVariables, |
|
|
) |
|
|
from typing_extensions import NotRequired, TypedDict |
|
|
|
|
|
from agents.util._types import MaybeAwaitable |
|
|
|
|
|
from .exceptions import UserError |
|
|
from .run_context import RunContextWrapper |
|
|
|
|
|
if TYPE_CHECKING: |
|
|
from .agent import Agent |
|
|
|
|
|
|
|
|
class Prompt(TypedDict): |
|
|
"""Prompt configuration to use for interacting with an OpenAI model.""" |
|
|
|
|
|
id: str |
|
|
"""The unique ID of the prompt.""" |
|
|
|
|
|
version: NotRequired[str] |
|
|
"""Optional version of the prompt.""" |
|
|
|
|
|
variables: NotRequired[dict[str, ResponsesPromptVariables]] |
|
|
"""Optional variables to substitute into the prompt.""" |
|
|
|
|
|
|
|
|
@dataclass |
|
|
class GenerateDynamicPromptData: |
|
|
"""Inputs to a function that allows you to dynamically generate a prompt.""" |
|
|
|
|
|
context: RunContextWrapper[Any] |
|
|
"""The run context.""" |
|
|
|
|
|
agent: Agent[Any] |
|
|
"""The agent for which the prompt is being generated.""" |
|
|
|
|
|
|
|
|
DynamicPromptFunction = Callable[[GenerateDynamicPromptData], MaybeAwaitable[Prompt]] |
|
|
"""A function that dynamically generates a prompt.""" |
|
|
|
|
|
|
|
|
class PromptUtil: |
|
|
@staticmethod |
|
|
async def to_model_input( |
|
|
prompt: Prompt | DynamicPromptFunction | None, |
|
|
context: RunContextWrapper[Any], |
|
|
agent: Agent[Any], |
|
|
) -> ResponsePromptParam | None: |
|
|
if prompt is None: |
|
|
return None |
|
|
|
|
|
resolved_prompt: Prompt |
|
|
if isinstance(prompt, dict): |
|
|
resolved_prompt = prompt |
|
|
else: |
|
|
func_result = prompt(GenerateDynamicPromptData(context=context, agent=agent)) |
|
|
if inspect.isawaitable(func_result): |
|
|
resolved_prompt = await func_result |
|
|
else: |
|
|
resolved_prompt = func_result |
|
|
if not isinstance(resolved_prompt, dict): |
|
|
raise UserError("Dynamic prompt function must return a Prompt") |
|
|
|
|
|
return { |
|
|
"id": resolved_prompt["id"], |
|
|
"version": resolved_prompt.get("version"), |
|
|
"variables": resolved_prompt.get("variables"), |
|
|
} |
|
|
|
