functional

Class info

🛈 DocStrings

High-level functional interfaces for LLMling agent.

auto_callable

auto_callable(
    model: str | Model | KnownModelName,
    *,
    system_prompt: str | None = None,
    retries: int = 3,
) -> Callable[
    [Callable[P, Coroutine[Any, Any, R]]], Callable[P, Coroutine[Any, Any, R]]
]

auto_callable(
    model: str | Model | KnownModelName,
    *,
    system_prompt: str | None = None,
    retries: int = 3,
) -> Callable[[Callable[P, R]], Callable[P, Coroutine[Any, Any, R]]]

auto_callable(
    model: str | Model | KnownModelName,
    *,
    system_prompt: str | None = None,
    retries: int = 3,
) -> AnyCallable

Use function signature as schema for LLM responses.

This decorator uses the function's: - Type hints - Docstring - Parameter names and defaults as a schema for getting structured responses from the LLM.

Parameters:

Name	Type	Description	Default
model	str | Model | KnownModelName	Model to use for responses	required
system_prompt	str | None	Optional system instructions	None
retries	int	Max retries for failed responses	3

Example

@auto_callable("gpt-5") async def analyze_sentiment(text: str) -> dict[str, float]: '''Analyze sentiment scores (positive/negative) for text.'''

Source code in src/llmling_agent/functional/auto_generate.py

def auto_callable[R, **P](
    model: str | models.Model | models.KnownModelName,
    *,
    system_prompt: str | None = None,
    retries: int = 3,
) -> AnyCallable:
    """Use function signature as schema for LLM responses.

    This decorator uses the function's:
    - Type hints
    - Docstring
    - Parameter names and defaults
    as a schema for getting structured responses from the LLM.

    Args:
        model: Model to use for responses
        system_prompt: Optional system instructions
        retries: Max retries for failed responses

    Example:
        @auto_callable("gpt-5")
        async def analyze_sentiment(text: str) -> dict[str, float]:
            '''Analyze sentiment scores (positive/negative) for text.'''
    """

    def decorator(
        func: Callable[P, R] | Callable[P, Coroutine[Any, Any, R]],
    ) -> Callable[P, Coroutine[Any, Any, R]]:
        # Get function info once
        sig = inspect.signature(func)
        doc = inspect.getdoc(func) or ""

        @functools.wraps(func)
        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            # Bind arguments to create context
            bound = sig.bind(*args, **kwargs)
            bound.apply_defaults()
            arg_values = dict(bound.arguments)

            # Create prompt from signature and args
            prompt = (
                f"Based on this function:\n\n"
                f"def {func.__name__}{sig!s}:\n"
                f'    """{doc}"""\n\n'
                f"Generate response for inputs: {arg_values}"
            )

            # Use get_structured to get typed response
            return_type = sig.return_annotation
            if return_type is inspect.Parameter.empty:
                return_type = str  # type: ignore[assignment]

            return await get_structured(
                prompt=prompt,
                response_type=return_type,  # type: ignore[arg-type]
                model=model,
                system_prompt=system_prompt,
                max_retries=retries,
            )

        return wrapper

    return decorator

get_structured async

get_structured(
    prompt: str,
    response_type: type[T],
    model: ModelType,
    *,
    system_prompt: str | None = None,
    max_retries: int = 3,
    error_handler: Callable[[Exception], T | None] | None = None,
) -> T

Get structured output from LLM using function calling.

This function creates a temporary agent that uses the class constructor as a tool to generate structured output. It handles: - Type conversion from Python types to JSON schema - Constructor parameter validation - Error handling with optional recovery

Parameters:

Name	Type	Description	Default
prompt	str	The prompt to send to the LLM	required
response_type	type[T]	The type to create (class with typed constructor)	required
model	ModelType	model to use	required
system_prompt	str | None	Optional system instructions	None
max_retries	int	Max attempts for parsing (default: 3)	3
error_handler	Callable[[Exception], T | None] | None	Optional error handler for recovery	None

Returns:

Type	Description
T	Instance of response_type

Example

class TaskResult:
    '''Analysis result for a task.'''
    def __init__(
        self,
        success: bool,
        message: str,
        due_date: datetime | None = None
    ):
        self.success = success
        self.message = message
        self.due_date = due_date

result = await get_structured(
    "Analyze task: Deploy monitoring",
    TaskResult,
    system_prompt="You analyze task success"
)
print(f"Success: {result.success}")

Raises:

Type	Description
TypeError	If response_type is not a valid type
ValueError	If constructor schema cannot be created
Exception	If LLM call fails and no error_handler recovers

Source code in src/llmling_agent/functional/structure.py

async def get_structured[T](
    prompt: str,
    response_type: type[T],
    model: ModelType,
    *,
    system_prompt: str | None = None,
    max_retries: int = 3,
    error_handler: Callable[[Exception], T | None] | None = None,
) -> T:
    """Get structured output from LLM using function calling.

    This function creates a temporary agent that uses the class constructor
    as a tool to generate structured output. It handles:
    - Type conversion from Python types to JSON schema
    - Constructor parameter validation
    - Error handling with optional recovery

    Args:
        prompt: The prompt to send to the LLM
        response_type: The type to create (class with typed constructor)
        model: model to use
        system_prompt: Optional system instructions
        max_retries: Max attempts for parsing (default: 3)
        error_handler: Optional error handler for recovery

    Returns:
        Instance of response_type

    Example:
        ```python
        class TaskResult:
            '''Analysis result for a task.'''
            def __init__(
                self,
                success: bool,
                message: str,
                due_date: datetime | None = None
            ):
                self.success = success
                self.message = message
                self.due_date = due_date

        result = await get_structured(
            "Analyze task: Deploy monitoring",
            TaskResult,
            system_prompt="You analyze task success"
        )
        print(f"Success: {result.success}")
        ```

    Raises:
        TypeError: If response_type is not a valid type
        ValueError: If constructor schema cannot be created
        Exception: If LLM call fails and no error_handler recovers
    """
    """Get structured output from LLM using function calling."""
    async with Agent(
        model=model,
        system_prompt=system_prompt or [],
        name="structured",
        retries=max_retries,
    ) as agent:
        try:
            return await agent.talk.extract(prompt, response_type)
        except Exception as e:
            if error_handler and (err_result := error_handler(e)):
                return err_result
            raise

get_structured_multiple async

get_structured_multiple(prompt: str, target: type[T], model: ModelType) -> list[T]

Extract multiple structured instances from text.

Source code in src/llmling_agent/functional/structure.py

async def get_structured_multiple[T](
    prompt: str,
    target: type[T],
    model: ModelType,
) -> list[T]:
    """Extract multiple structured instances from text."""
    async with Agent(model=model, name="structured") as agent:
        return await agent.talk.extract_multiple(prompt, target)

pick_one async

pick_one(prompt: str, options: type[T | Enum] | list[T], model: ModelType) -> T

Pick one option from a list of choices.

Source code in src/llmling_agent/functional/structure.py

async def pick_one[T](
    prompt: str,
    options: type[T | Enum] | list[T],
    model: ModelType,
) -> T:
    """Pick one option from a list of choices."""
    instances: dict[str, T] = {}

    # Create mapping and descriptions
    if isinstance(options, type):
        if issubclass(options, Enum):
            choices = {e.name: (e.value, str(e.value)) for e in options}
        else:
            literal_opts = get_args(options)
            choices = {str(opt): (opt, str(opt)) for opt in literal_opts}
    else:  # List
        choices = {str(i): (opt, repr(opt)) for i, opt in enumerate(options)}

    async def select_option(option: Literal[tuple(choices.keys())]) -> str:  # type: ignore
        """Pick one of the available options.

        Args:
            option: Which option to pick
        """
        instances["selected"] = choices[option][0]
        return f"Selected: {option}"

    # Add options to docstring
    docs = "\n".join(f"- {k}: {desc}" for k, (_, desc) in choices.items())
    assert select_option.__doc__
    select_option.__doc__ += f"\nOptions:\n{docs}"
    sys_prompt = "Select the most appropriate option based on the context."
    async with Agent(model=model, system_prompt=sys_prompt) as agent:
        agent.tools.register_tool(select_option, enabled=True)
        await agent.run(prompt)
        return instances["selected"]

run_agent async

run_agent(
    prompt: AnyPromptType | Image | PathLike[str],
    image_url: str | None = None,
    *,
    result_type: type[TResult],
    **kwargs: Unpack[AgentKwargs],
) -> TResult

run_agent(
    prompt: AnyPromptType | Image | PathLike[str],
    image_url: str | None = None,
    **kwargs: Unpack[AgentKwargs],
) -> str

run_agent(
    prompt: AnyPromptType | Image | PathLike[str],
    image_url: str | None = None,
    *,
    result_type: type[Any] | None = None,
    **kwargs: Unpack[AgentKwargs],
) -> Any

Run prompt through agent and return result.

Source code in src/llmling_agent/functional/run.py

async def run_agent(
    prompt: AnyPromptType | PIL.Image.Image | os.PathLike[str],
    image_url: str | None = None,
    *,
    result_type: type[Any] | None = None,
    **kwargs: Unpack[AgentKwargs],
) -> Any:
    """Run prompt through agent and return result."""
    async with Agent[Any](**kwargs) as agent:
        if image_url:
            image = ImageURLContent(url=image_url)
            result = await agent.run(prompt, image, result_type=result_type)
        else:
            result = await agent.run(prompt, result_type=result_type)
        return result.content

run_agent_sync

run_agent_sync(
    prompt: AnyPromptType | Image | PathLike[str],
    image_url: str | None = None,
    *,
    result_type: type[TResult],
    **kwargs: Unpack[AgentKwargs],
) -> TResult

run_agent_sync(
    prompt: AnyPromptType | Image | PathLike[str],
    image_url: str | None = None,
    **kwargs: Unpack[AgentKwargs],
) -> str

run_agent_sync(
    prompt: AnyPromptType | Image | PathLike[str],
    image_url: str | None = None,
    *,
    result_type: type[Any] | None = None,
    **kwargs: Unpack[AgentKwargs],
) -> Any

Sync wrapper for run_agent.

Source code in src/llmling_agent/functional/run.py

def run_agent_sync(
    prompt: AnyPromptType | PIL.Image.Image | os.PathLike[str],
    image_url: str | None = None,
    *,
    result_type: type[Any] | None = None,
    **kwargs: Unpack[AgentKwargs],
) -> Any:
    """Sync wrapper for run_agent."""
    coro = run_agent(prompt, image_url, result_type=result_type, **kwargs)  # type: ignore
    return asyncio.run(coro)