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

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 to ensure type safety across the API boundary:

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 mirrors this structure exactly:

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 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:

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 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:

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) that maps agent keys to their implementation functions. The translator builds a lookup table to resolve React Flow node IDs into executable Python functions:

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:

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:

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:

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:

# 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:

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. This TypedDict enables message passing and data accumulation across the workflow:

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:

{
  "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

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

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.
• Backend translation: The create_graph function in 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, 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. 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}