LangGraph on AsyncFlow directly¶

A LangGraph application can run its tools, services and nodes as AsyncFlow tasks without Flowgentic. This recipe is the port that the substitution experiment made of the agentic demo: examples/ai-hpc-coupling/direct_agentic_application.py keeps the LangGraph graph, node bodies, tools, state and policy guards of agentic_application.py, and replaces each Flowgentic decorator with a direct AsyncFlow call. The planner's node also writes by hand the acceptance policy that the Flowgentic version gets from accept; see Acceptance. From Flowgentic it imports only Rejected, which the demo's shared checks and decision model raise.

tests/integration/test_substitution.py runs both versions with the same arguments and finds the same decisions and results, in CI.

1. Create and shut down the engine¶

The application creates the WorkflowEngine on its backends and registers its shutdown. The demo's runner uses an AsyncExitStack:

flow = await WorkflowEngine.create(backend=backends)
stack.push_async_callback(flow.shutdown)
application = await build_direct_agentic_campaign(
    flow,
    settings,
    decision_model,
)

backends is the same list of named local backends as in the Flowgentic version (compute and ai).

2. Services and tasks¶

flow.function_task registers a task on a named backend; service=True makes it a resident service. Calling a service task returns its future, and awaiting the future gives the service:

# 1. The same resident surrogate service as the deterministic demo --------
@flow.function_task(service=True, backend="ai")
async def load_surrogate() -> SurrogateService:
    return await SurrogateService.load()

service_future = load_surrogate()
service = await service_future

The demo's reasoning steps are plain @flow.function_task(backend="ai") tasks.

3. Tools and nodes¶

Two small helpers do what Flowgentic's AGENT_TOOL_AS_FUNCTION and EXECUTION_BLOCK decorators did. asyncflow_tool wraps an AsyncFlow task in a LangChain tool, with an optional retry; asyncflow_block runs a graph node as an AsyncFlow block:

# 8. What the Flowgentic decorators did, written directly against AsyncFlow ---
def asyncflow_tool(
    description: str,
    *,
    retry: Callable[[Callable[[], Awaitable[Any]]], Awaitable[Any]] | None = None,
) -> Callable[[Callable[..., Any]], BaseTool]:
    """Expose an AsyncFlow task as a LangChain tool, with an optional retry."""

    def decorate(task: Callable[..., Any]) -> BaseTool:
        @wraps(task)
        async def invoke(*args: Any, **kwargs: Any) -> Any:
            if retry is None:
                return await task(*args, **kwargs)
            return await retry(lambda: task(*args, **kwargs))

        return tool(invoke, description=description)

    return decorate


def asyncflow_block(
    flow: WorkflowEngine,
    node: Callable[[Any], Awaitable[dict[str, Any]]],
) -> Callable[[Any], Awaitable[dict[str, Any]]]:
    """Run a LangGraph node as an AsyncFlow block."""
    block = flow.block(node)

    @wraps(node)
    async def run_block(state: Any) -> dict[str, Any]:
        return await block(state)

    return run_block

The tools stack asyncflow_tool on top of flow.function_task:

# 2. Bounded scientific capabilities exposed as agent tools ---------------
@asyncflow_tool("Query the resident surrogate for a bounded candidate proposal.")
@flow.function_task(backend="ai")
async def query_surrogate(
    center: float,
    radius: float,
    cycle: int,
) -> dict[str, Any]:
    """Return the surrogate's candidate proposal for one campaign cycle."""
    return await service.propose(center, radius, settings.batch_size, cycle)

@asyncflow_tool(
    "Run one approved scientific simulation on compute.",
    retry=retry_transient_simulation,
)
@flow.function_task(backend="compute")
async def run_simulation(
    x: float,
    cycle: int,
    candidate_index: int,
    fail_once: bool = False,
) -> dict[str, Any]:
    """Evaluate one policy-approved candidate."""
    return await evaluate_candidate(x, cycle, candidate_index, fail_once)

@asyncflow_tool("Assimilate simulation evidence into the resident surrogate.")
@flow.function_task(backend="ai")
async def update_surrogate(
    results: list[dict[str, Any]],
    radius: float,
    cycle: int,
) -> dict[str, Any]:
    """Update the surrogate with an approved simulation batch."""
    return await service.assimilate(results, radius, cycle)

and the graph's nodes are AsyncFlow blocks:

# 7. The scientific loop is unchanged; only node implementation differs ---
graph = StateGraph(AgenticCampaignState)
nodes = {
    "planner_agent": asyncflow_block(flow, planner_agent),
    "simulation_executor": asyncflow_block(flow, simulation_executor),
    "analyst_agent": asyncflow_block(flow, analyst_agent),
    "supervisor_agent": asyncflow_block(flow, supervisor_agent),
}
for name, node in nodes.items():
    graph.add_node(name, node)
graph.add_edge(START, "planner_agent")
graph.add_edge("planner_agent", "simulation_executor")
graph.add_edge("simulation_executor", "analyst_agent")
graph.add_edge("analyst_agent", "supervisor_agent")
graph.add_edge("supervisor_agent", END)

The same helper serves a model that selects a tool: the substitution tests give asyncflow_tool(description)(flow.function_task(run_simulation, backend="compute")) to LangGraph's ToolNode, and it behaves as the Flowgentic tool does (test_llm_selected_tool_runs_as_an_asyncflow_task).

4. Retries¶

AsyncFlow 0.6.0 has no per-task retry or timeout, so the application writes the policy it needs. The demo retries a simulation once after a transient failure, with 5 s per attempt, and passes the policy to asyncflow_tool as retry=retry_transient_simulation:

async def retry_transient_simulation(
    operation: Callable[[], Awaitable[ResultType]],
) -> ResultType:
    """Apply the baseline's deterministic two-attempt transient-failure policy."""
    try:
        return await asyncio.wait_for(operation(), timeout=5.0)
    except TransientSimulationError:
        await asyncio.sleep(0.05)
        return await asyncio.wait_for(operation(), timeout=5.0)

Nothing else is retried: graph nodes run once, with no timeout. The Flowgentic version uses the same helper; see Retries and timeouts.

5. Handoffs¶

A handoff tool runs an AsyncFlow task that returns a LangGraph Command. The direct branch of this test helper is the whole recipe, 17 lines against Flowgentic's one call:

def handoff_tool(
    integration: str, flow: Any, adapter: Any, agent_name: str
) -> BaseTool:
    """A tool that hands the task to another agent in the parent graph."""
    description = f"Hand the task to the {agent_name}."
    if integration == "flowgentic":
        return adapter.execution_wrappers.create_task_description_handoff_tool(
            agent_name, description
        )

    # Direct: the same Command as Flowgentic's tool, built in an AsyncFlow task.
    @flow.function_task
    async def hand_off(task_description: str, state: Any) -> Command:
        base_state = state if isinstance(state, dict) else state.model_dump()
        agent_input = {
            **base_state,
            "messages": [HumanMessage(content=task_description)],
        }
        return Command(goto=[Send(agent_name, agent_input)], graph=Command.PARENT)

    @tool(f"transfer_to_{agent_name}", description=description)
    async def transfer(
        task_description: Annotated[str, "What the next agent should do."],
        state: Annotated[Any, InjectedState],
    ) -> Command:
        return await hand_off(task_description, state)

    return transfer

Run it¶

uv run --frozen --extra examples examples/ai-hpc-coupling/run_agentic_demo.sh application --integration direct

The run prints the same five cycles and proof points as the Flowgentic version (see Getting started), and its agentic_demo_results/campaign_summary.json records "integration": "direct". In this repository the direct run still loads Flowgentic, through the shared runner and demo support code; a standalone direct application needs only radical-asyncflow, langgraph and langchain-core.

What changes without Flowgentic¶

	With Flowgentic	Directly
Engine	Created and shut down by `LangraphIntegration`	Created and shut down by the application
Tools, tasks, services, nodes	`asyncflow` decorator	`flow.function_task`, `flow.block`, `langchain_core.tools.tool`
Retry and timeout	Only what the application writes	Only what the application writes
Handoff tool	`create_task_description_handoff_tool`	About 17 lines
Accepting the planner's proposals	`accept`, with spans and evaluation events	The same policy in a hand-written loop, 3 more code lines, with no record in the trace
Third-party packages (fresh resolve, Python 3.10)	74	42