Function resolve_magic_input

quantmind/magic.py:62–98

Parse ``natural_language`` into ``(input_obj, cfg_obj)`` for ``target_flow``. Args: natural_language: User-supplied free-form description of intent. target_flow: The flow function to resolve for. Must accept ``input`` (positional) and ``cfg`` (keyword) parameters.

(
    natural_language: str,
    *,
    target_flow: Callable[..., Awaitable[Any]],
    resolver_model: str = "gpt-4o-mini",
    resolver_instructions: str | None = None,
)

  with the description's content.

Input schema:
{input_schema}

Cfg schema:
{cfg_schema}
"""


async def resolve_magic_input(
    natural_language: str,
    *,
    target_flow: Callable[..., Awaitable[Any]],
    resolver_model: str = "gpt-5.6-luna",
    resolver_instructions: str | None = None,
) -> tuple[Any, Any]:
    """Parse ``natural_language`` into ``(input_obj, cfg_obj)`` for ``target_flow``.

    Args:
        natural_language: User-supplied free-form description of intent.
        target_flow: The flow function to resolve for. Must accept
            ``input`` (positional) and ``cfg`` (keyword) parameters.
        resolver_model: LLM used by the resolver agent.
        resolver_instructions: Optional override for the resolver's
            system prompt template. Receives ``flow_name``,
            ``input_schema``, and ``cfg_schema`` via ``str.format``.

    Returns:
        Tuple of ``(input_obj, cfg_obj)`` populated by the resolver.
    """
    input_type, cfg_type = _introspect_flow_signature(target_flow)
    template = resolver_instructions or _RESOLVER_INSTRUCTIONS
    instructions = template.format(
        flow_name=target_flow.__name__,
        input_schema=_pydantic_schema_str(input_type),
        cfg_schema=_pydantic_schema_str(cfg_type),
    )
    resolver: Agent[Any] = Agent(
        name=f"magic_resolver_{target_flow.__name__}",
        instructions=instructions,
        model=resolver_model,