LangGraph on AsyncFlow with Flowgentic

LangraphIntegration runs a LangGraph graph's tools, services, tasks and nodes as AsyncFlow tasks. It opens an AsyncFlow WorkflowEngine, and its asyncflow decorator turns an async function into a LangChain tool, an AsyncFlow task, a resident service or a graph node.

Frozen in maintenance mode

The execution adapter gets no new features. The substitution experiment ran the same application with and without it. Without it, the application kept the same results, task identity, cancellation, service reuse and backend placement, in about the same code. Up to 0.3, the adapter also wrapped every call and graph node in an implicit retry and timeout, which re-ran scientific work; 0.4.0 removes them. For new applications, see LangGraph on AsyncFlow directly.

The code on this page comes from the agentic demo, examples/ai-hpc-coupling/agentic_application.py, and from tests/integration/test_substitution.py. Both run in CI: the demo through tests/integration/test_ai_hpc_agentic_demo.py and the substitution tests.

Open the integration

LangraphIntegration(backend=...) takes one AsyncFlow backend or a list of named backends. The demo creates two local backends, compute and ai:

async def create_local_backends(compute_workers: int) -> list[LocalExecutionBackend]:
    """Create laptop-safe stand-ins for RHAPSODY AI and compute backends."""
    compute = await LocalExecutionBackend(
        ThreadPoolExecutor(max_workers=compute_workers), name="compute"
    )
    ai = await LocalExecutionBackend(ThreadPoolExecutor(max_workers=2), name="ai")
    return [compute, ai]

Use the integration as an async context manager. Entering it creates the WorkflowEngine; leaving it shuts the engine down. The demo's runner enters it on an AsyncExitStack:

adapter = await stack.enter_async_context(
    LangraphIntegration(backend=backends)
)
flow = adapter.flow
application = await build_agentic_campaign(
    adapter,
    settings,
    decision_model,
)

Inside the context, the integration exposes:

Attribute	Content
flow	The AsyncFlow WorkflowEngine
execution_wrappers	The asyncflow decorator and the handoff tool

The integration always creates its own engine: it cannot run on an engine the application has already created.

Decorate functions

integration.execution_wrappers.asyncflow(func=None, *, flow_type, backend=None, capture_stdio=False, tool_description=None) registers an async function with AsyncFlow. flow_type is required in practice: without it the decorator fails with AttributeError. Any other keyword raises TypeError. The demo binds it to a short name:

flowgentic = integration.execution_wrappers.asyncflow

flow_type selects what the decorated function becomes:

AsyncFlowType	AsyncFlow call	The decorated function	Use it for
AGENT_TOOL_AS_FUNCTION	flow.function_task	A LangChain tool whose call awaits the task's result	A tool that an agent or node code calls
FUNCTION_TASK	flow.function_task	An async function that awaits the task's result	A task called from node code
SERVICE_TASK	flow.function_task(..., service=True)	An async function that returns the task's future; awaiting the future gives the service	A resident service, such as a loaded model
EXECUTION_BLOCK	flow.block	An async function of the graph state	A graph node run as an AsyncFlow block

backend places a task on a named backend; EXECUTION_BLOCK does not pass it to AsyncFlow. tool_description is the tool's description for the model. Each call runs once, with no retry and no timeout; see Retries and timeouts.

A resident service

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

service_future = await load_surrogate()
service = await service_future

The first await returns the service task's future; the second returns the service. See Services.

Agent tools

# 2. Bounded scientific capabilities exposed as agent tools ---------------
@flowgentic(
    flow_type=AsyncFlowType.AGENT_TOOL_AS_FUNCTION,
    backend="ai",
    tool_description=(
        "Query the resident surrogate for a bounded candidate proposal."
    ),
)
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)

@flowgentic(
    flow_type=AsyncFlowType.AGENT_TOOL_AS_FUNCTION,
    backend="compute",
    tool_description="Run one approved scientific simulation on 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)

@flowgentic(
    flow_type=AsyncFlowType.AGENT_TOOL_AS_FUNCTION,
    backend="ai",
    tool_description="Assimilate simulation evidence into the resident surrogate.",
)
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)

run_simulation runs on compute, the other tools on ai. The executor below retries a transient simulation failure once.

Tasks

# Agent reasoning is independently placeable on the named AI backend.
@flowgentic(flow_type=AsyncFlowType.FUNCTION_TASK, backend="ai")
async def reason_about_plan(context: dict[str, Any]) -> dict[str, Any]:
    started = time.perf_counter()
    decision = await decision_model.plan(context)
    return {
        "decision": decision.model_dump(),
        "event": timed_event(
            name="agent.planner",
            backend="ai",
            cycle=int(context["cycle"]),
            started=started,
        ),
    }

Graph nodes

# 7. The scientific loop is unchanged; only node implementation differs ---
graph = StateGraph(AgenticCampaignState)
nodes = {
    "planner_agent": flowgentic(flow_type=AsyncFlowType.EXECUTION_BLOCK)(
        planner_agent
    ),
    "simulation_executor": flowgentic(flow_type=AsyncFlowType.EXECUTION_BLOCK)(
        simulation_executor
    ),
    "analyst_agent": flowgentic(flow_type=AsyncFlowType.EXECUTION_BLOCK)(
        analyst_agent
    ),
    "supervisor_agent": flowgentic(flow_type=AsyncFlowType.EXECUTION_BLOCK)(
        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)

With OpenTelemetry configured, each node is an AsyncFlow block span and each tool call an execute_tool span, with the tool's task under it; see Telemetry.

Calling tools from node code

The demo's nodes call their tools directly with .ainvoke, so the tools run as AsyncFlow tasks without an LLM choosing them. The executor runs one simulation per approved candidate, concurrently:

# 4. Deterministic executor: invoke approved simulation tools concurrently -
# The application retries a transient simulation failure once.
async def simulation_executor(
    state: AgenticCampaignState,
) -> dict[str, Any]:
    results = await asyncio.gather(
        *(
            retry_transient_simulation(
                lambda x=x, index=index: run_simulation.ainvoke(
                    {
                        "x": x,
                        "cycle": state["cycle"],
                        "candidate_index": index,
                        "fail_once": (
                            settings.inject_failure
                            and state["cycle"] == 1
                            and index == 0
                        ),
                    }
                )
            )
            for index, x in enumerate(state["candidates"])
        )
    )
    return {
        "results": results,
        "spent": state["spent"] + len(results),
        "trace": [*state["trace"], *(item["event"] for item in results)],
        "messages": [deterministic_executor_message(len(results))],
    }

Let a model select a tool

An AGENT_TOOL_AS_FUNCTION tool also works in LangGraph's ToolNode, where the model chooses it. The substitution tests build this graph:

def agent_graph(
    model: GenericFakeChatModel,
    tools: list[BaseTool],
    state_schema: Any = BaseLLMAgentState,
) -> CompiledStateGraph:
    """A model node that calls tools through ToolNode until it answers."""

    async def call_model(state: Any) -> dict[str, Any]:
        messages = state["messages"] if isinstance(state, dict) else state.messages
        return {"messages": [await model.ainvoke(messages)]}

    graph = StateGraph(state_schema)
    graph.add_node("model", call_model)
    graph.add_node("tools", ToolNode(tools))
    graph.add_edge(START, "model")
    graph.add_conditional_edges("model", tools_condition)
    graph.add_edge("tools", "model")
    return graph.compile()

and give it the demo's simulation as a tool, through Flowgentic or directly:

def science_tool(integration: str, flow: Any, adapter: Any) -> BaseTool:
    """The demo's simulation, exposed as an agent tool on the compute backend."""

    async def run_simulation(x: float) -> dict[str, float]:
        """Evaluate one candidate with the demo's synthetic simulation."""
        result = await demo("campaign_common").evaluate_candidate(x, 0, 0)
        return {"x": result["x"], "value": result["value"]}

    description = "Run one bounded scientific simulation on compute."
    if integration == "flowgentic":
        return adapter.execution_wrappers.asyncflow(
            flow_type=AsyncFlowType.AGENT_TOOL_AS_FUNCTION,
            backend="compute",
            tool_description=description,
        )(run_simulation)
    direct = demo("direct_agentic_application")
    return direct.asyncflow_tool(description)(
        flow.function_task(run_simulation, backend="compute")
    )

With a scripted model that calls run_simulation, the tool runs as one AsyncFlow task on compute and returns its result to the model (test_llm_selected_tool_runs_as_an_asyncflow_task).

LangGraph 1.x's ToolNode re-raises tool errors by default, where 0.6 returned them to the model as messages. A tool that fails aborts the graph run, unless the application passes handle_tool_errors=True to ToolNode.

Hand off to another agent

integration.execution_wrappers.create_task_description_handoff_tool(agent_name, description) returns a tool named transfer_to_<agent_name>. When a supervisor agent calls it with a task description, an AsyncFlow task builds Command(goto=[Send(agent_name, ...)], graph=Command.PARENT), which sends the graph state, with the task description as its only message, to the agent_name node of the parent graph. The graph state can be a Pydantic model or a TypedDict:

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

test_supervisor_hands_off_to_a_worker runs the handoff from a supervisor subgraph to a researcher node, for both kinds of state. From October 2025 until 0.3.0, this tool failed on every call on main.

Chat models

flowgentic.utils.llm_providers.ChatLLMProvider(provider, model=..., ...) builds a LangChain chat model:

provider	Model class	Credentials
"openrouter"	ChatOpenRouter, a ChatOpenAI pointed at https://openrouter.ai/api/v1	api_key=, or the OPEN_ROUTER_API_KEY environment variable
"chatopenai" or "chatgpt"	ChatOpenAI	As for langchain-openai
"ollama"	ChatOllama	None

The provider name is case-insensitive, and other keyword arguments go to the model class. tests/unit/test_llm_providers.py builds each one offline.

Things to know

• Flowgentic retries nothing. A graph node takes LangGraph's retry_policy and timeout, and a task call is wrapped by the application; see Retries and timeouts.
• Importing Flowgentic configures no logging. flowgentic.configure_logging() prints its records; see Logging.
• Project status lists what is not validated, such as remote and HPC backends.