run_nodes

Class info

🛈 DocStrings

Automatic agent function execution.

run_nodes

run_nodes(config: JoinablePathLike | AgentsManifest, **kwargs: Any) -> dict[str, Any]

Run node functions synchronously.

Convenience wrapper around run_nodes_async for sync contexts. See run_nodes_async for full documentation.

Parameters:

Name	Type	Description	Default
config	JoinablePathLike | AgentsManifest	Agent configuration path	required
**kwargs	Any	Arguments to pass to run_nodes_async	{}

Returns:

Type	Description
dict[str, Any]	Dict mapping function names to their results

Source code in src/llmling_agent/running/run_nodes.py

def run_nodes(config: JoinablePathLike | AgentsManifest, **kwargs: Any) -> dict[str, Any]:
    """Run node functions synchronously.

    Convenience wrapper around run_nodes_async for sync contexts.
    See run_nodes_async for full documentation.

    Args:
        config: Agent configuration path
        **kwargs: Arguments to pass to run_nodes_async

    Returns:
        Dict mapping function names to their results
    """
    return asyncio.run(run_nodes_async(config, **kwargs))

run_nodes_async async

run_nodes_async(
    config: JoinablePathLike | AgentsManifest,
    *,
    module: str | None = None,
    functions: list[str] | None = None,
    inputs: dict[str, Any] | None = None,
    parallel: bool = False,
) -> dict[str, Any]

Execute node functions with dependency handling.

Parameters:

Name	Type	Description	Default
config	JoinablePathLike | AgentsManifest	Node configuration (path or manifest)	required
module	str | None	Optional module to discover functions from	None
functions	list[str] | None	Optional list of function names to run (auto-discovers if None)	None
inputs	dict[str, Any] | None	Optional input values for function parameters	None
parallel	bool	Whether to run independent functions in parallel	False

Returns:

Type	Description
dict[str, Any]	Dict mapping function names to their results

Example

@node_function
async def analyze(analyzer: Agent) -> str:
    return await analyzer.run("...")

results = await run_nodes_async("agents.yml")
print(results["analyze"])

Source code in src/llmling_agent/running/run_nodes.py

async def run_nodes_async(
    config: JoinablePathLike | AgentsManifest,
    *,
    module: str | None = None,
    functions: list[str] | None = None,
    inputs: dict[str, Any] | None = None,
    parallel: bool = False,
) -> dict[str, Any]:
    """Execute node functions with dependency handling.

    Args:
        config: Node configuration (path or manifest)
        module: Optional module to discover functions from
        functions: Optional list of function names to run (auto-discovers if None)
        inputs: Optional input values for function parameters
        parallel: Whether to run independent functions in parallel

    Returns:
        Dict mapping function names to their results

    Example:
        ```python
        @node_function
        async def analyze(analyzer: Agent) -> str:
            return await analyzer.run("...")

        results = await run_nodes_async("agents.yml")
        print(results["analyze"])
        ```
    """
    from llmling_agent import AgentPool

    # Find functions to run
    if module:
        discovered = discover_functions(module)
    else:
        # Use calling module
        import inspect

        frame = inspect.currentframe()
        while frame:
            if frame.f_globals.get("__name__") != __name__:
                break
            frame = frame.f_back
        if not frame:
            msg = "Could not determine calling module"
            raise RuntimeError(msg)
        discovered = discover_functions(frame.f_globals["__file__"])

    if functions:
        discovered = [f for f in discovered if f.name in functions]

    # Run with pool
    async with AgentPool[None](config) as pool:
        return await execute_functions(
            discovered,
            pool,
            inputs=inputs,
            parallel=parallel,
        )