rigging.prompt

Treat empty function signatures as prompts for structured chat interfaces.

DEFAULT_DOC = 'Convert the following inputs to outputs ({func_name}).' module-attribute

Default docstring if none is provided to a prompt function.

DEFAULT_MAX_ROUNDS = 3 module-attribute

Default maximum number of rounds for a prompt to run until outputs are parsed.

Ctx(tag: str | None = None, prefix: str | None = None, example: str | Model | None = None) dataclass

Used in type annotations to provide additional context for the prompt construction.

You can use this annotation on inputs and ouputs to prompt functions.

tag_override = Annotated[str, Ctx(tag="custom_tag", ...)]

Prompt(func: t.Callable[P, t.Coroutine[t.Any, t.Any, R]], attempt_recovery: bool = True, drop_dialog: bool = True, max_rounds: int = DEFAULT_MAX_ROUNDS, inputs: list[Input] = list(), output: Output = lambda: ChatOutput(id='chat', context=Ctx())(), watch_callbacks: list[WatchChatCallback] = list(), params: GenerateParams | None = None, _generator_id: str | None = None, _generator: Generator | None = None, _pipeline: ChatPipeline | None = None) dataclass

Bases: Generic[P, R]

Prompts wrap hollow functions and create structured chat interfaces for passing inputs into a ChatPipeline and parsing outputs.

attempt_recovery: bool = True class-attribute instance-attribute

Whether the prompt should attempt to recover from errors in output parsing.

docstring: str property

The docstring for the prompt function.

drop_dialog: bool = True class-attribute instance-attribute

When attempting recovery, whether to drop intermediate dialog while parsing was being resolved.

func: t.Callable[P, t.Coroutine[t.Any, t.Any, R]] instance-attribute

The function that the prompt wraps. This function should be a coroutine.

inputs: list[Input] = dataclasses.field(default_factory=list) class-attribute instance-attribute

The structured input handlers for the prompt.

max_rounds: int = DEFAULT_MAX_ROUNDS class-attribute instance-attribute

The maximum number of rounds the prompt should try to reparse outputs.

output: Output = dataclasses.field(default_factory=lambda: ChatOutput(id='chat', context=Ctx())) class-attribute instance-attribute

The structured output handler for the prompt.

params: GenerateParams | None = None class-attribute instance-attribute

The parameters to be used when generating chats for this prompt.

pipeline: ChatPipeline | None property

If available, the resolved Chat Pipeline for the prompt.

template: str property

The dynamic jinja2 template for the prompt function.

watch_callbacks: list[WatchChatCallback] = dataclasses.field(default_factory=list) class-attribute instance-attribute

Callbacks to be passed any chats produced while executing this prompt.

bind(other: ChatPipeline | Generator | Chat | str) -> t.Callable[P, t.Coroutine[t.Any, t.Any, R]]

Binds the prompt to a pipeline, generator, or chat and returns a scoped run callable.

@rg.prompt
def say_hello(name: str) -> str:
    """Say hello to {{ name }}"""

await say_hello.bind("gpt-3.5-turbo")("the world")

Parameters:

• other (ChatPipeline | Generator | Chat | str) –

  The pipeline, generator, generator id, or chat to bind to.

Returns:

• Callable[P, Coroutine[Any, Any, R]] –

  A callable for executing this prompt

Source code in rigging/prompt.py

def bind(self, other: ChatPipeline | Generator | Chat | str) -> t.Callable[P, t.Coroutine[t.Any, t.Any, R]]:
    """
    Binds the prompt to a pipeline, generator, or chat and returns a scoped run callable.

    ```
    @rg.prompt
    def say_hello(name: str) -> str:
        \"""Say hello to {{ name }}\"""

    await say_hello.bind("gpt-3.5-turbo")("the world")
    ```

    Args:
        other: The pipeline, generator, generator id, or chat to bind to.

    Returns:
        A callable for executing this prompt
    """
    pipeline = self._resolve_to_pipeline(other)
    if pipeline.on_failed == "skip":
        raise NotImplementedError(
            "pipeline.on_failed='skip' cannot be used for prompt methods that return one object"
        )

    async def run(*args: P.args, **kwargs: P.kwargs) -> R:
        results = await self.bind_many(pipeline)(1, *args, **kwargs)
        return results[0]

    return run

bind_many(other: ChatPipeline | Generator | Chat | str) -> t.Callable[Concatenate[int, P], t.Coroutine[t.Any, t.Any, list[R]]]

Binds the prompt to a pipeline, generator, or chat and returns a scoped run_many callable.

@rg.prompt
def say_hello(name: str) -> str:
    """Say hello to {{ name }}"""

await say_hello.bind("gpt-3.5-turbo")(5, "the world")

Parameters:

• other (ChatPipeline | Generator | Chat | str) –

  The pipeline, generator, generator id, or chat to bind to.

Returns:

• Callable[Concatenate[int, P], Coroutine[Any, Any, list[R]]] –

  A callable for executing this prompt.

Source code in rigging/prompt.py

def bind_many(
    self, other: ChatPipeline | Generator | Chat | str
) -> t.Callable[Concatenate[int, P], t.Coroutine[t.Any, t.Any, list[R]]]:
    """
    Binds the prompt to a pipeline, generator, or chat and returns a scoped run_many callable.

    ```
    @rg.prompt
    def say_hello(name: str) -> str:
        \"""Say hello to {{ name }}\"""

    await say_hello.bind("gpt-3.5-turbo")(5, "the world")
    ```

    Args:
        other: The pipeline, generator, generator id, or chat to bind to.

    Returns:
        A callable for executing this prompt.
    """
    pipeline = self._resolve_to_pipeline(other)
    if pipeline.on_failed == "include" and not isinstance(self.output, ChatOutput):
        raise NotImplementedError("pipeline.on_failed='include' cannot be used with prompts that process outputs")

    async def run_many(count: int, /, *args: P.args, **kwargs: P.kwargs) -> list[R]:
        content = self.render(*args, **kwargs)
        _pipeline = (
            pipeline.fork(content)
            .until(
                self._until_parsed,
                attempt_recovery=self.attempt_recovery,
                drop_dialog=self.drop_dialog,
                max_rounds=self.max_rounds,
            )
            .with_(self.params)
        )
        chats = await _pipeline.run_many(count)

        coros = [watch(chats) for watch in self.watch_callbacks if watch not in pipeline.watch_callbacks]
        await asyncio.gather(*coros)

        return [self.process(chat) for chat in chats]

    return run_many

bind_over(other: ChatPipeline | Generator | Chat | str | None = None) -> t.Callable[Concatenate[t.Sequence[Generator | str], P], t.Coroutine[t.Any, t.Any, list[R]]]

Binds the prompt to a pipeline, generator, or chat and returns a scoped run_over callable.

@rg.prompt
def say_hello(name: str) -> str:
    """Say hello to {{ name }}"""

await say_hello.bind("gpt-3.5-turbo")(["gpt-4o", "gpt-4"], "the world")

Parameters:

• other (ChatPipeline | Generator | Chat | str | None, default: None ) –

  The pipeline, generator, generator id, or chat to bind to.

Returns:

• Callable[Concatenate[Sequence[Generator | str], P], Coroutine[Any, Any, list[R]]] –

  A callable for executing this prompt.

Source code in rigging/prompt.py

def bind_over(
    self, other: ChatPipeline | Generator | Chat | str | None = None
) -> t.Callable[Concatenate[t.Sequence[Generator | str], P], t.Coroutine[t.Any, t.Any, list[R]]]:
    """
    Binds the prompt to a pipeline, generator, or chat and returns a scoped run_over callable.

    ```
    @rg.prompt
    def say_hello(name: str) -> str:
        \"""Say hello to {{ name }}\"""

    await say_hello.bind("gpt-3.5-turbo")(["gpt-4o", "gpt-4"], "the world")
    ```

    Args:
        other: The pipeline, generator, generator id, or chat to bind to.

    Returns:
        A callable for executing this prompt.
    """
    include_original = other is not None

    if other is None:
        pipeline = get_generator("base!base").chat().catch(on_failed="skip")  # TODO: Clean this up
    else:
        pipeline = self._resolve_to_pipeline(other)

    if pipeline.on_failed == "include" and not isinstance(self.output, ChatOutput):
        raise NotImplementedError("pipeline.on_failed='include' cannot be used with prompts that process outputs")

    async def run_over(generators: t.Sequence[Generator | str], /, *args: P.args, **kwargs: P.kwargs) -> list[R]:
        content = self.render(*args, **kwargs)
        _pipeline = (
            pipeline.fork(content)
            .until(
                self._until_parsed,
                attempt_recovery=self.attempt_recovery,
                drop_dialog=self.drop_dialog,
                max_rounds=self.max_rounds,
            )
            .with_(self.params)
        )
        chats = await _pipeline.run_over(*generators, include_original=include_original)

        coros = [watch(chats) for watch in self.watch_callbacks if watch not in pipeline.watch_callbacks]
        await asyncio.gather(*coros)

        return [self.process(chat) for chat in chats]

    return run_over

clone(*, skip_callbacks: bool = False) -> Prompt[P, R]

Creates a deep copy of this prompt.

Parameters:

• skip_callbacks (bool, default: False ) –

  Whether to skip copying the watch callbacks.

Returns:

• Prompt[P, R] –

  A new instance of the prompt.

Source code in rigging/prompt.py

def clone(self, *, skip_callbacks: bool = False) -> Prompt[P, R]:
    """
    Creates a deep copy of this prompt.

    Args:
        skip_callbacks: Whether to skip copying the watch callbacks.

    Returns:
        A new instance of the prompt.
    """
    new = Prompt(
        func=self.func,
        _pipeline=self.pipeline,
        params=self.params.model_copy() if self.params is not None else None,
        attempt_recovery=self.attempt_recovery,
        drop_dialog=self.drop_dialog,
        max_rounds=self.max_rounds,
    )
    if not skip_callbacks:
        new.watch_callbacks = self.watch_callbacks.copy()
    return new

process(chat: Chat) -> R

Attempt to parse the output from a chat into the expected return type.

Source code in rigging/prompt.py

def process(self, chat: Chat) -> R:
    """
    Attempt to parse the output from a chat into the expected return type.
    """
    return self.output.from_chat(chat)  # type: ignore

render(*args: P.args, **kwargs: P.kwargs) -> str

Pass the arguments to the jinja2 template and render the full prompt.

Source code in rigging/prompt.py

def render(self, *args: P.args, **kwargs: P.kwargs) -> str:
    """
    Pass the arguments to the jinja2 template and render the full prompt.
    """
    env = Environment(
        trim_blocks=True,
        lstrip_blocks=True,
        keep_trailing_newline=True,
        undefined=StrictUndefined,
    )
    jinja_template = env.from_string(self.template)

    signature = inspect.signature(self.func)
    bound_args = signature.bind(*args, **kwargs)
    bound_args.apply_defaults()

    for input_ in self.inputs:
        bound_args.arguments[to_snake(input_.tag)] = input_.to_xml(bound_args.arguments[input_.name])

    return jinja_template.render(**bound_args.arguments)

run(*args: P.args, **kwargs: P.kwargs) -> R async

Use the prompt to run the function with the provided arguments and return the output.

Parameters:

• *args (args, default: () ) –

  The positional arguments for the prompt function.

• **kwargs (kwargs, default: {} ) –

  The keyword arguments for the prompt function.

Returns:

• R –

  The output of the prompt function.

Source code in rigging/prompt.py

async def run(self, *args: P.args, **kwargs: P.kwargs) -> R:
    """
    Use the prompt to run the function with the provided arguments and return the output.

    Args:
        *args: The positional arguments for the prompt function.
        **kwargs: The keyword arguments for the prompt function.

    Returns:
        The output of the prompt function.
    """
    if self.pipeline is None:
        raise RuntimeError(
            "Prompt cannot be executed as a standalone function without being assigned a pipeline or generator"
        )
    return await self.bind(self.pipeline)(*args, **kwargs)

run_many(count: int, /, *args: P.args, **kwargs: P.kwargs) -> list[R] async

Use the prompt to run the function multiple times with the provided arguments and return the output.

Parameters:

• count (int) –

  The number of times to run the prompt.

• *args (args, default: () ) –

  The positional arguments for the prompt function.

• **kwargs (kwargs, default: {} ) –

  The keyword arguments for the prompt function.

Returns:

• list[R] –

  The outputs of the prompt function.

Source code in rigging/prompt.py

async def run_many(self, count: int, /, *args: P.args, **kwargs: P.kwargs) -> list[R]:
    """
    Use the prompt to run the function multiple times with the provided arguments and return the output.

    Args:
        count: The number of times to run the prompt.
        *args: The positional arguments for the prompt function.
        **kwargs: The keyword arguments for the prompt function.

    Returns:
        The outputs of the prompt function.
    """
    if self.pipeline is None:
        raise RuntimeError(
            "Prompt cannot be executed as a standalone function without being assigned a pipeline or generator"
        )
    return await self.bind_many(self.pipeline)(count, *args, **kwargs)

run_over(generators: t.Sequence[Generator | str], /, *args: P.args, **kwargs: P.kwargs) -> list[R] async

Executes the prompt process across multiple generators.

For each generator, a pipeline is created and the generator is replaced before the run call. All callbacks and parameters are preserved.

If this prompt has a pipeline assigned, it will be included in the run.

Warning

The implementation currently skips any failed chats and only processes successful chats. This may change in the future.

Parameters:

• generators (Sequence[Generator | str]) –

  A sequence of generators to be used for the generation process.

Returns:

• list[R] –

  A list of generatated Chats.

Source code in rigging/prompt.py

async def run_over(self, generators: t.Sequence[Generator | str], /, *args: P.args, **kwargs: P.kwargs) -> list[R]:
    """
    Executes the prompt process across multiple generators.

    For each generator, a pipeline is created and the generator is replaced
    before the run call. All callbacks and parameters are preserved.

    If this prompt has a pipeline assigned, it will be included in the run.

    Warning:
        The implementation currently skips any failed chats and only
        processes successful chats. This may change in the future.

    Parameters:
        generators: A sequence of generators to be used for the generation process.

    Returns:
        A list of generatated Chats.
    """
    return await self.bind_over(self.pipeline)(generators, *args, **kwargs)

set_(attempt_recovery: bool | None = None, drop_dialog: bool | None = None, max_rounds: int | None = None) -> Prompt[P, R]

Helper to allow updates to the parsing configuration.

Parameters:

• attempt_recovery (bool | None, default: None ) –

  Whether the prompt should attempt to recover from errors in output parsing.

• drop_dialog (bool | None, default: None ) –

  When attempting recovery, whether to drop intermediate dialog while parsing was being resolved.

• max_rounds (int | None, default: None ) –

  The maximum number of rounds the prompt should try to reparse outputs.

Returns:

• Prompt[P, R] –

  Self

Source code in rigging/prompt.py

def set_(
    self, attempt_recovery: bool | None = None, drop_dialog: bool | None = None, max_rounds: int | None = None
) -> Prompt[P, R]:
    """
    Helper to allow updates to the parsing configuration.

    Args:
        attempt_recovery: Whether the prompt should attempt to recover from errors in output parsing.
        drop_dialog: When attempting recovery, whether to drop intermediate dialog while parsing was being resolved.
        max_rounds: The maximum number of rounds the prompt should try to reparse outputs.

    Returns:
        Self
    """
    self.attempt_recovery = attempt_recovery or self.attempt_recovery
    self.drop_dialog = drop_dialog or self.drop_dialog
    self.max_rounds = max_rounds or self.max_rounds
    return self

watch(*callbacks: WatchChatCallback) -> Prompt[P, R]

Registers a callback to monitor any chats produced for this prompt

Parameters:

• *callbacks (WatchChatCallback, default: () ) –

  The callback functions to be executed.

async def log(chats: list[Chat]) -> None:
    ...

@rg.prompt()
async def summarize(text: str) -> str:
    ...

summarize.watch(log)(...)

or

async def log(chats: list[Chat]) -> None:
    ...

async def _summarize(text: str) -> str:
    ...

summarize = rg.prompt(_summarize).watch(log)

Returns:

• Prompt[P, R] –

  Self

Source code in rigging/prompt.py

def watch(self, *callbacks: WatchChatCallback) -> Prompt[P, R]:
    """
    Registers a callback to monitor any chats produced for this prompt

    Args:
        *callbacks: The callback functions to be executed.

    ```
    async def log(chats: list[Chat]) -> None:
        ...

    @rg.prompt()
    async def summarize(text: str) -> str:
        ...

    summarize.watch(log)(...)
    ```
    or
    ```
    async def log(chats: list[Chat]) -> None:
        ...

    async def _summarize(text: str) -> str:
        ...

    summarize = rg.prompt(_summarize).watch(log)
    ```

    Returns:
        Self
    """
    for callback in callbacks:
        if callback not in self.watch_callbacks:
            self.watch_callbacks.append(callback)
    return self

with_(params: t.Optional[GenerateParams] = None, **kwargs: t.Any) -> Prompt[P, R]

Assign specific generation parameter overloads for this prompt.

Parameters:

• params (Optional[GenerateParams], default: None ) –

  The parameters to set for the underlying chat pipeline.

• **kwargs (Any, default: {} ) –

  An alternative way to pass parameters as keyword arguments.

Returns:

• Prompt[P, R] –

  Self

Source code in rigging/prompt.py

def with_(self, params: t.Optional[GenerateParams] = None, **kwargs: t.Any) -> Prompt[P, R]:
    """
    Assign specific generation parameter overloads for this prompt.

    Args:
        params: The parameters to set for the underlying chat pipeline.
        **kwargs: An alternative way to pass parameters as keyword arguments.

    Returns:
        Self
    """
    self.params = params if params is not None else GenerateParams(**kwargs)
    return self

prompt(func: t.Callable[P, t.Coroutine[t.Any, t.Any, R]] | t.Callable[P, R] | None = None, /, *, pipeline: ChatPipeline | None = None, generator: Generator | None = None, generator_id: str | None = None) -> t.Callable[[t.Callable[P, t.Coroutine[t.Any, t.Any, R]] | t.Callable[P, R]], Prompt[P, R]] | Prompt[P, R]

Convert a hollow function into a Prompt, which can be called directly or passed a chat pipeline to execute the function and parse the outputs.

``` from dataclasses import dataclass import rigging as rg

@dataclass class ExplainedJoke: chat: rg.Chat setup: str punchline: str explanation: str

@rg.prompt(generator_id="gpt-3.5-turbo") async def write_joke(topic: str) -> ExplainedJoke: """Write a joke.""" ...

await write_joke("programming")

Note: A docstring is not required, but this can be used to provide guidance to the model, or even handle any number of input transormations. Any input parameter which is not handled inside the docstring will be automatically added and formatted internally.

Note: Output parameters can be basic types, dataclasses, rigging models, lists, or tuples. Internal inspection will attempt to ensure your output types are valid, but there is no guarantee of complete coverage/safety. It's recommended to check rigging.prompt.Prompt.template to inspect the generated jinja2 template.

Note: If you annotate the return value of the function as a rigging.chat.Chat object, then no output parsing will take place and you can parse objects out manually.

You can also use Chat in any number of type annotation inside tuples or dataclasses.
All instances will be filled with the final chat object transparently.

Note: All input parameters and output types can be annotated with the rigging.prompt.Ctx annotation to provide additional context for the prompt. This can be used to override the xml tag, provide a prefix string, or example content which will be placed inside output xml tags.

In the case of output parameters, especially in tuples, you might have xml tag collisions
between the same basic types. Manually annotating xml tags with [rigging.prompt.Ctx][] is
recommended.

Args: func: The function to convert into a prompt. pipeline: An optional pipeline to use for the prompt. generator: An optional generator to use for the prompt. generator_id: An optional generator id to use for the prompt.

Returns: A prompt instance or a function that can be used to create a prompt.

Source code in rigging/prompt.py

def prompt(
    func: t.Callable[P, t.Coroutine[t.Any, t.Any, R]] | t.Callable[P, R] | None = None,
    /,
    *,
    pipeline: ChatPipeline | None = None,
    generator: Generator | None = None,
    generator_id: str | None = None,
) -> t.Callable[[t.Callable[P, t.Coroutine[t.Any, t.Any, R]] | t.Callable[P, R]], Prompt[P, R]] | Prompt[P, R]:
    """
    Convert a hollow function into a Prompt, which can be called directly or passed a
    chat pipeline to execute the function and parse the outputs.

    ```
    from dataclasses import dataclass
    import rigging as rg

    @dataclass
    class ExplainedJoke:
        chat: rg.Chat
        setup: str
        punchline: str
        explanation: str

    @rg.prompt(generator_id="gpt-3.5-turbo")
    async def write_joke(topic: str) -> ExplainedJoke:
        \"""Write a joke.\"""
        ...

    await write_joke("programming")

    Note:
        A docstring is not required, but this can be used to provide guidance to the model, or
        even handle any number of input transormations. Any input parameter which is not
        handled inside the docstring will be automatically added and formatted internally.

    Note:
        Output parameters can be basic types, dataclasses, rigging models, lists, or tuples.
        Internal inspection will attempt to ensure your output types are valid, but there is
        no guarantee of complete coverage/safety. It's recommended to check
        [rigging.prompt.Prompt.template][] to inspect the generated jinja2 template.

    Note:
        If you annotate the return value of the function as a [rigging.chat.Chat][] object,
        then no output parsing will take place and you can parse objects out manually.

        You can also use Chat in any number of type annotation inside tuples or dataclasses.
        All instances will be filled with the final chat object transparently.

    Note:
        All input parameters and output types can be annotated with the [rigging.prompt.Ctx][] annotation
        to provide additional context for the prompt. This can be used to override the xml tag, provide
        a prefix string, or example content which will be placed inside output xml tags.

        In the case of output parameters, especially in tuples, you might have xml tag collisions
        between the same basic types. Manually annotating xml tags with [rigging.prompt.Ctx][] is
        recommended.

    Args:
        func: The function to convert into a prompt.
        pipeline: An optional pipeline to use for the prompt.
        generator: An optional generator to use for the prompt.
        generator_id: An optional generator id to use for the prompt.

    Returns:
        A prompt instance or a function that can be used to create a prompt.
    """
    if sum(arg is not None for arg in (pipeline, generator, generator_id)) > 1:
        raise ValueError("Only one of pipeline, generator, or generator_id can be provided")

    def make_prompt(func: t.Callable[P, t.Coroutine[t.Any, t.Any, R]] | t.Callable[P, R]) -> Prompt[P, R]:
        return Prompt[P, R](
            func=func,  # type: ignore
            _generator_id=generator_id,
            _pipeline=pipeline,
            _generator=generator,
        )

    if func is not None:
        return make_prompt(func)
    return make_prompt