core/graph/ — LangGraph Workflow Orchestration

What is this & why does it exist?

The graph module is the nervous system of the hedge fund. AgentState is a TypedDict with three fields — messages, data, and metadata — that every agent reads from and writes to. The merge_dicts reducer handles concurrent writes from parallel agents. The create_workflow() function in main.py builds a LangGraph StateGraph by connecting start_node → analyst agents (fan-out) → risk_manager → portfolio_manager → END. Although only ~180 lines, this is the most critical code — it defines how agents compose.

Why Is It Designed This Way?

LangGraph’s StateGraph was chosen over a hand-rolled orchestrator because it provides built-in support for parallel fan-out (all analyst agents run concurrently), typed state channels with reducer functions (so concurrent writes merge cleanly), and a compile-to-graph execution model that enables visualization and debugging. The AgentState TypedDict pattern gives type safety without runtime overhead — every agent declares exactly what it reads and writes. The merge_dicts reducer was chosen over operator.or_ because it preserves all keys from both dicts even when nested keys overlap.

AgentState & merge_dicts (graph/state.py)

from typing_extensions import Annotated, Sequence, TypedDict
import operator
from langchain_core.messages import BaseMessage

def merge_dicts(a: dict[str, any], b: dict[str, any]) -> dict[str, any]:
    # WHY: Shallow merge ensures parallel agents don't overwrite each other's keys.
    # operator.or_ (|) would work for flat dicts but merge_dicts is explicit and safe.
    return {**a, **b}

class AgentState(TypedDict):
    # WHY: Annotated channels tell LangGraph HOW to merge concurrent writes.
    messages: Annotated[Sequence[BaseMessage], operator.add]  # append-only
    data: Annotated[dict[str, any], merge_dicts]              # shallow merge
    metadata: Annotated[dict[str, any], merge_dicts]          # shallow merge

create_workflow() (main.py)

def create_workflow(selected_analysts=None):
    """Create the workflow with selected analysts."""
    workflow = StateGraph(AgentState)
    workflow.add_node("start_node", start)

    analyst_nodes = get_analyst_nodes()

    # WHY: Default to all analysts when none selected (CLI passed empty list)
    if selected_analysts is None:
        selected_analysts = list(analyst_nodes.keys())

    # WHY: Dynamic wiring - only add nodes the user actually selected
    for analyst_key in selected_analysts:
        node_name, node_func = analyst_nodes[analyst_key]
        workflow.add_node(node_name, node_func)
        workflow.add_edge("start_node", node_name)  # fan-out from start

    # WHY: Risk and portfolio management are always present (non-optional)
    workflow.add_node("risk_management_agent", risk_management_agent)
    workflow.add_node("portfolio_manager", portfolio_management_agent)

    # WHY: Fan-in - every analyst must complete before risk assessment
    for analyst_key in selected_analysts:
        node_name = analyst_nodes[analyst_key][0]
        workflow.add_edge(node_name, "risk_management_agent")

    workflow.add_edge("risk_management_agent", "portfolio_manager")
    workflow.add_edge("portfolio_manager", END)
    workflow.set_entry_point("start_node")
    return workflow