Sequential pattern

In a sequential (pipeline) graph, each stage runs after the previous one and passes its results on through the graph state. Conditional edges send the run to an error handler when a stage fails.

The example is examples/langgraph-integration/design_patterns/sequential/research_agent/: a research pipeline with five stages and an error handler. Two stages are LLM agents with tools; the others are deterministic AsyncFlow tasks.

flowchart LR
  P[preprocess] --> R[research_agent] --> C[context_preparation] --> S[synthesis_agent] --> F[finalize_output]
  P -.-> E[error_handler]
  R -.-> E
  C -.-> E
  S -.-> E

Run it

The agents use OpenRouter: set OPEN_ROUTER_API_KEY, then run make examples-sequential-research. CI, which has no credentials, runs the example with a scripted chat model in place of OpenRouter (tests/integration/test_design_patterns_offline.py); tests/integration/test_integration.py runs it live when FLOWGENTIC_RUN_LLM_INTEGRATION=1 is set.

The graph

Every node is an EXECUTION_BLOCK, and each transition is a conditional edge:

def build_workflow(self) -> StateGraph:
    """Build and return the complete workflow graph."""
    # Register all tools first
    self.tools_registry._register_toolkit()

    # Create state graph
    workflow = StateGraph(WorkflowState)

    # Add all nodes
    all_nodes = self.nodes.get_all_nodes()
    for node_name, node_function in all_nodes.items():
        workflow.add_node(node_name, node_function)

    # Add conditional edges
    workflow.add_conditional_edges(
        "preprocess",
        self.edges.should_continue_after_preprocessing,
        {"research_agent": "research_agent", "error_handler": "error_handler"},
    )

    workflow.add_conditional_edges(
        "research_agent",
        self.edges.should_continue_after_research,
        {
            "context_preparation": "context_preparation",
            "error_handler": "error_handler",
        },
    )

    workflow.add_conditional_edges(
        "context_preparation",
        self.edges.should_continue_after_context,
        {"synthesis_agent": "synthesis_agent", "error_handler": "error_handler"},
    )

    workflow.add_conditional_edges(
        "synthesis_agent",
        self.edges.should_continue_after_synthesis,
        {"finalize_output": "finalize_output", "error_handler": "error_handler"},
    )

    # Add edges to END
    workflow.add_edge("finalize_output", END)
    workflow.add_edge("error_handler", END)

    # Set entry point
    workflow.set_entry_point("preprocess")

    return workflow

An edge function reads the stage's result from the state:

@staticmethod
def should_continue_after_research(state: WorkflowState) -> str:
    """Determine next step after research agent."""
    if state.research_agent_output and state.research_agent_output.success:
        return "context_preparation"
    else:
        return "error_handler"

An agent stage

The research stage builds an agent with langchain.agents.create_agent, gives it two tools, and records its output in the state. A failure is appended to state.errors and recorded as success=False in the stage's output, which the next edge routes to the error handler:

@property
def research_agent_node(self):
    @self.agents_manager.execution_wrappers.asyncflow(
        flow_type=AsyncFlowType.EXECUTION_BLOCK
    )
    async def _research_agent_node(state: WorkflowState) -> WorkflowState:
        """Research agent execution node."""
        print("🔍 Research Agent Node: Starting research and analysis...")

        try:
            start_time = asyncio.get_event_loop().time()

            tools = [
                self.tools_registry.get_tool_by_name("web_search"),
                self.tools_registry.get_tool_by_name("data_analysis"),
            ]

            research_agent = create_agent(
                model=ChatLLMProvider(
                    provider="OpenRouter", model="google/gemini-2.5-flash"
                ),
                tools=tools,
            )

            research_state = {
                "messages": [
                    SystemMessage(
                        content="You are a research agent specializing in technology analysis. Your job is to gather comprehensive information, analyze data, and provide detailed insights. Always use your tools to get the most current and accurate information. Dont do more than 3 queries. Every time you want to invoke tool, explain your planning planning strategy beforehand"
                    ),
                    HumanMessage(content=state.user_input),
                ]
            }
            research_result = await research_agent.ainvoke(research_state)
            execution_time = asyncio.get_event_loop().time() - start_time

            if "messages" in research_result and isinstance(
                research_result["messages"], list
            ):
                state.messages.extend(research_result["messages"])

            agent_output = AgentOutput(
                agent_name="Research Agent",
                output_content=research_result["messages"][-1].content,
                execution_time=execution_time,
                tools_used=["web_search_tool", "data_analysis_tool"],
                success=True,
            )

            state.research_agent_output = agent_output
            state.current_stage = "research_complete"

            print(f"✅ Research Agent complete in {execution_time:.2f}s")

            return state

        except Exception as e:
            error_msg = f"Research agent error: {str(e)}"
            state.errors.append(error_msg)
            state.research_agent_output = AgentOutput(
                agent_name="Research Agent",
                output_content="",
                execution_time=0,
                success=False,
                error_message=error_msg,
            )
            state.current_stage = "research_failed"

        return state

    return _research_agent_node

Its tools are AGENT_TOOL_AS_FUNCTION functions, so each tool call the model makes runs as an AsyncFlow task:

@self.agents_manager.execution_wrappers.asyncflow(
    flow_type=AsyncFlowType.AGENT_TOOL_AS_FUNCTION
)
async def web_search_tool(query: str) -> str:
    """Search the web for information."""
    await asyncio.sleep(1)  # Simulate network delay
    return f"Search results for '{query}': Found relevant information about renewable energy storage, including battery technologies, grid integration, and market trends."

The example's tools return canned results after a short sleep; they stand in for real search and analysis services.

Running the graph

main.py opens the integration on a local backend, builds the graph and compiles it with an in-memory checkpointer:

async def start_app():
    backend = await LocalExecutionBackend(ThreadPoolExecutor())

    async with LangraphIntegration(backend=backend) as agents_manager:
        # Build workflow
        workflow_builder = WorkflowBuilder(agents_manager)
        workflow = workflow_builder.build_workflow()

        # Compile the app
        memory = InMemorySaver()
        app = workflow.compile(checkpointer=memory)

It then streams the run and returns the final state. A run that ends in the error handler raises RuntimeError with the stages' errors, so it exits non-zero.