# How React Flow Graphs Translate into LangGraph Execution in AI Hedge Fund

> Discover how the ai-hedge-fund system transforms React Flow graphs into LangGraph execution. Learn about node and edge array transmission, FastAPI backend integration, and dynamic StateGraph construction for AI workflows.

- Repository: [Virat Singh/ai-hedge-fund](https://github.com/virattt/ai-hedge-fund)
- Tags: architecture
- Published: 2026-03-09

---

**The virattt/ai-hedge-fund system converts visual React Flow diagrams into executable LangGraph workflows by transmitting node and edge arrays to a FastAPI backend, which dynamically constructs a StateGraph with automatic risk-management node injection and parallel agent orchestration.**

The virattt/ai-hedge-fund project provides a visual workflow builder where users drag and drop analyst, portfolio-manager, and risk-manager nodes using React Flow. When the user triggers execution, the frontend serializes the canvas state into structured arrays that the backend transforms into a LangGraph execution plan. This translation layer bridges visual programming with LangChain's graph-based agent orchestration, enabling complex hedge-fund simulations to run from a simple UI interaction.

## From Canvas to Code: The React Flow Payload Structure

The React Flow canvas exports its state as two primary arrays that capture the visual topology. The frontend defines these interfaces in [`app/frontend/src/services/types.ts`](https://github.com/virattt/ai-hedge-fund/blob/main/app/frontend/src/services/types.ts) to ensure type safety across the API boundary:

```typescript
export interface GraphNode {
  id: string;
  type?: string;
  data?: any;
  position?: { x: number; y: number };
}

export interface GraphEdge {
  id: string;
  source: string;
  target: string;
}

```

When the user clicks **Run**, the frontend assembles a `HedgeFundRequest` payload containing these arrays along with portfolio parameters. The backend schema in [`app/backend/models/schemas.py`](https://github.com/virattt/ai-hedge-fund/blob/main/app/backend/models/schemas.py) mirrors this structure exactly:

```python
class BaseHedgeFundRequest(BaseModel):
    tickers: List[str]
    graph_nodes: List[GraphNode]      # Receives React Flow node definitions

    graph_edges: List[GraphEdge]      # Receives connection topology

    ...

```

This bidirectional contract ensures that the visual graph structure transmits losslessly from the TypeScript frontend to the Python backend.

## The FastAPI Entry Point and Graph Compilation

The `/hedge-fund/run` endpoint in [`app/backend/routes/hedge_fund.py`](https://github.com/virattt/ai-hedge-fund/blob/main/app/backend/routes/hedge_fund.py) receives the payload and immediately delegates graph construction to the graph service. The route extracts the node and edge arrays and triggers the translation logic:

```python
graph = create_graph(
    graph_nodes=request_data.graph_nodes,
    graph_edges=request_data.graph_edges,
)
graph = graph.compile()

```

The `create_graph` function in [`app/backend/services/graph.py`](https://github.com/virattt/ai-hedge-fund/blob/main/app/backend/services/graph.py) contains the core translation engine that converts the React Flow abstraction into a LangChain `StateGraph`. This process involves ten distinct construction phases that validate, expand, and wire the execution topology.

## Constructing the LangGraph StateGraph

### Initializing the Graph Container

The translation begins by instantiating a `StateGraph` with a shared `AgentState` type. The code adds a mandatory entry node that serves as the unified starting point for all execution paths:

```python
graph = StateGraph(AgentState)
graph.add_node("start_node", start)  # Fixed entry point

```

### Resolving Agent Configurations

The system maintains an `ANALYST_CONFIG` registry (defined in [`src/utils/analysts.py`](https://github.com/virattt/ai-hedge-fund/blob/main/src/utils/analysts.py)) that maps agent keys to their implementation functions. The translator builds a lookup table to resolve React Flow node IDs into executable Python functions:

```python
analyst_nodes = {key: (f"{key}_agent", config["agent_func"]) 
                 for key, config in ANALYST_CONFIG.items()}

```

### Node Classification and ID Extraction

The translator extracts all node IDs from the incoming `graph_nodes` array and categorizes them by type. It uses `extract_base_agent_key` to strip random suffixes (like `_abc123`) from IDs to identify **portfolio_manager** nodes separately from analyst nodes:

```python
agent_ids = [node.id for node in graph_nodes]
portfolio_manager_nodes = [
    node_id for node_id in agent_ids 
    if extract_base_agent_key(node_id) == "portfolio_manager"
]

```

### Dynamic Agent Function Creation

For each valid analyst node, the system generates a unique agent function using `create_agent_function`. This factory binds the specific analyst configuration to a LangChain runnable, which is then registered as a node in the StateGraph:

```python
for agent_id in agent_ids:
    if agent_id not in portfolio_manager_nodes:
        agent_func = create_agent_function(analyst_key)
        graph.add_node(unique_agent_id, agent_function)

```

### Risk Management Node Injection

A critical architectural requirement mandates that all portfolio decisions pass through risk validation. For every portfolio-manager node detected, the translator automatically generates a paired risk-management node with a matching suffix:

```python
risk_manager_id = f"risk_management_agent_{suffix}"
graph.add_node(portfolio_manager_id, portfolio_manager_agent)
graph.add_node(risk_manager_id, risk_management_agent)

```

This injection happens transparently during graph construction, ensuring compliance checks execute even if the user did not manually place a risk node on the canvas.

### Edge Wiring and Topology Rewriting

The translator iterates through the `graph_edges` array to establish execution dependencies. It distinguishes between direct analyst-to-analyst connections and analyst-to-portfolio-manager connections. The latter are intercepted and rerouted through the corresponding risk-management node:

```python

# Analyst → Portfolio Manager edges are rewritten to go through Risk Manager

for edge in graph_edges:
    if is_portfolio_target(edge.target):
        direct_to_portfolio_managers.append((edge.source, edge.target))
    else:
        graph.add_edge(edge.source, edge.target)

```

### Connecting Orphaned Nodes and Finalization

Any node without incoming edges receives an automatic connection from `start_node` to ensure execution entry. Finally, all portfolio-manager nodes link to the `END` terminator, and the entry point is explicitly set:

```python
graph.add_edge("start_node", orphaned_node_id)  # For nodes without parents

graph.add_edge(risk_manager_id, portfolio_manager_id)
graph.set_entry_point("start_node")

```

The returned `StateGraph` object is then compiled into an execution plan ready for invocation.

## Shared State Management Across Agents

All nodes in the constructed graph operate on a common `AgentState` defined in [`src/graph/state.py`](https://github.com/virattt/ai-hedge-fund/blob/main/src/graph/state.py). This TypedDict enables message passing and data accumulation across the workflow:

```python
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    data: Annotated[dict[str, any], merge_dicts]
    metadata: Annotated[dict[str, any], merge_dicts]

```

The `operator.add` annotation on `messages` enables automatic list concatenation as outputs propagate through the graph, while `merge_dicts` ensures that portfolio updates from multiple analysts consolidate into a unified state object.

## End-to-End Execution Flow

The complete translation and execution pipeline follows this sequence:

1. **User assembles a diagram** in React Flow, producing `graph_nodes` and `graph_edges` arrays.
2. **Frontend POST** to `/hedge-fund/run` includes the serialized topology.
3. **FastAPI route** extracts arrays and invokes `create_graph`.
4. **Graph service** builds a LangChain `StateGraph`, inserting hidden risk-manager nodes and rewiring edges.
5. **Compilation** via `graph.compile()` resolves the internal execution plan and dependency order.
6. **Execution** through `run_graph_async` walks the graph, invoking each agent with the shared `AgentState`.
7. **Results** stream back through the API as Server-Sent Events (SSE) to the frontend.

## Practical Implementation Examples

### Example 1: Minimal React Flow Payload

The following JSON represents a simple workflow with one analyst feeding a portfolio manager:

```json
{
  "graph_nodes": [
    { "id": "warren_buffett_1a2b3c", "type": "analyst", "data": { "name": "Warren Buffett" } },
    { "id": "portfolio_manager_4d5e6f", "type": "portfolio_manager" }
  ],
  "graph_edges": [
    { "id": "e1", "source": "warren_buffett_1a2b3c", "target": "portfolio_manager_4d5e6f" }
  ]
}

```

*Note: The backend will rewrite this edge to route through `risk_management_agent_4d5e6f` before reaching the portfolio manager.*

### Example 2: Translating to a LangGraph in Python

```python
from app.backend.services.graph import create_graph

# Simulating the payload from Example 1

nodes = [
    {"id": "warren_buffett_1a2b3c"},
    {"id": "portfolio_manager_4d5e6f"},
]
edges = [{"id": "e1", "source": "warren_buffett_1a2b3c", "target": "portfolio_manager_4d5e6f"}]

# Build and compile the StateGraph

lg = create_graph(graph_nodes=nodes, graph_edges=edges)
compiled = lg.compile()

# Conceptual execution topology:

# start_node → warren_buffett_1a2b3c → risk_management_agent_4d5e6f → portfolio_manager_4d5e6f → END

```

### Example 3: Executing the Compiled Graph

```python
from app.backend.services.graph import run_graph_async

result = await run_graph_async(
    graph=compiled,
    portfolio={"cash": 100_000, "positions": {}},
    tickers=["AAPL", "MSFT"],
    start_date="2024-01-01",
    end_date="2024-03-01",
    model_name="gpt-4o-mini",
    model_provider="OpenAI",
)

# Result contains final portfolio allocations and intermediate analyst signals

```

## Summary

- **React Flow serialization**: The frontend exports node and edge arrays via `GraphNode` and `GraphEdge` interfaces defined in [`app/frontend/src/services/types.ts`](https://github.com/virattt/ai-hedge-fund/blob/main/app/frontend/src/services/types.ts).
- **Backend translation**: The `create_graph` function in [`app/backend/services/graph.py`](https://github.com/virattt/ai-hedge-fund/blob/main/app/backend/services/graph.py) dynamically constructs a LangChain `StateGraph` from the visual topology.
- **Automatic risk injection**: Direct connections to portfolio managers are transparently rerouted through auto-generated `risk_management_agent_{suffix}` nodes.
- **Entry point handling**: Orphaned nodes without incoming edges automatically connect to the `start_node` entry point.
- **State sharing**: All agents operate on a shared `AgentState` TypedDict that accumulates messages and portfolio data across the execution graph.

## Frequently Asked Questions

### Does every portfolio manager connection automatically get a risk manager node?

Yes. The backend intercepts any edge targeting a portfolio-manager node and inserts a corresponding `risk_management_agent_{suffix}` node between the source analyst and the portfolio manager. This ensures all allocation decisions pass through risk validation before execution, regardless of the visual diagram's original wiring.

### What happens to orphaned nodes with no incoming edges?

During the wiring phase in [`app/backend/services/graph.py`](https://github.com/virattt/ai-hedge-fund/blob/main/app/backend/services/graph.py), the translator detects nodes that lack incoming connections and automatically attaches them to the `start_node` entry point. This guarantees that every agent executes exactly once when the graph begins, even if the user forgot to connect them in the React Flow canvas.

### Can custom analyst agents be added to the visual builder?

The system validates node types against `ANALYST_CONFIG` in [`src/utils/analysts.py`](https://github.com/virattt/ai-hedge-fund/blob/main/src/utils/analysts.py). Any analyst defined in this registry can be instantiated as a React Flow node and will resolve to its corresponding `agent_func` during graph construction. To add custom agents, you must register them in the configuration dictionary before the backend can recognize and instantiate their nodes.

### How does the frontend handle streaming results from LangGraph execution?

The compiled graph executes via `run_graph_async`, which yields results as they are produced by each node. The FastAPI endpoint streams these outputs back to the React frontend using Server-Sent Events (SSE), allowing the UI to display real-time analyst signals, risk management flags, and portfolio updates as the workflow progresses through the graph.