Main Pipeline Graph

Overview

Purpose: orchestrates the end-to-end NL2SQL pipeline as a LangGraph state machine.
Why it exists architecturally: central control plane that wires global nodes, subgraphs, and aggregation into a single execution DAG.
When it is invoked: via run_with_graph() runtime entry point.

Defining function: build_graph()
Source file path: packages/nl2sql/src/nl2sql/pipeline/graph.py

Boundary Definition

This graph is the top-level orchestrator. It encapsulates datasource resolution, decomposition (which builds the execution DAG), routing to subgraphs, aggregation, and answer synthesis. It does NOT implement subgraph internals (SQL planning/execution), per-node business logic, or adapter-specific execution; those are delegated to nodes and registered subgraphs.

Entry Conditions

Required state: - GraphState.user_query must be provided. - GraphState.trace_id is auto-generated if not provided.

Preconditions: - Runtime must be initialized with registries/services used by nodes (vector store, LLM registry, datasource registry, schema store, RBAC).

Triggering parent graph: - None. This is the root graph invoked by run_with_graph() in pipeline/runtime.py.

Exit Conditions

Successful completion criteria: - Reaches END after answer_synthesizer when aggregation and synthesis finish without fatal interruptions.

Failure exits: - datasource_resolver can short-circuit to END via resolver_route when no datasource is resolved/allowed, or when its answerability check refuses the question (QUESTION_NOT_ANSWERABLE); no other LLM call runs. - layer_router can return END if no execution_dag or layers are present. - layer_router returns END once every scan has run if none produced an artifact (every sub-query was denied or failed), and whenever the graph was built with execute=False. The aggregator and answer synthesizer are skipped, so the run's errors are the sub-queries' own -- for a policy denial, SECURITY_VIOLATION first -- and no LLM call is spent explaining an empty result. - layer_router raises PipelineExecutionError (an NL2SQLError carrying the PipelineError payload on .error) when no compatible subgraph exists for a datasource; this propagates as an exception from the router function. A conditional-edge router may only return routing decisions, so it cannot report the failure as a PipelineError value in state the way a node does.

Partial completion behavior: - If routing cannot schedule any scan-layer work but execution_dag exists and at least one scan produced an artifact, it routes to aggregator directly. If some scans produced artifacts and others did not, the aggregator still runs and reports the missing ones as AGGREGATOR_FAILED. - Errors added to GraphState.errors do not automatically stop the graph; the graph relies on routing logic and node behavior for termination. - aggregator_route ends the run when the aggregator failed (AggregatorResponse.errors is not empty), so the answer synthesizer never spends a model call on an empty result and AGGREGATOR_FAILED is the run's outcome.

Internal Node Composition

Execution order (nominal path): - curated_matcher — CuratedMatcherNode — packages/nl2sql/src/nl2sql/pipeline/nodes/curated_matcher/node.py — only when curated answers are configured and the run executes: answer with a verified query (curated_route → verified → layer_router), or fall back (generated → datasource_resolver). Without curated answers the node is not in the graph and datasource_resolver is the entry point. - datasource_resolver — DatasourceResolverNode — packages/nl2sql/src/nl2sql/pipeline/nodes/datasource_resolver/node.py — resolve candidate datasources, RBAC, and the answerability check (an LLM call). - decomposer — DecomposerNode — packages/nl2sql/src/nl2sql/pipeline/nodes/decomposer/node.py — decompose query into sub-queries, and build the ExecutionDAG from them (decomposer/dag.py). - layer_router — inline lambda + routes.build_scan_layer_router — packages/nl2sql/src/nl2sql/pipeline/routes.py — route each scan layer to a subgraph or aggregator. - <subgraph nodes> — via wrap_subgraph() — packages/nl2sql/src/nl2sql/pipeline/graph_utils.py — invoke registered subgraphs per scan node. - aggregator — EngineAggregatorNode — packages/nl2sql/src/nl2sql/pipeline/nodes/aggregator/node.py — execute aggregation DAG. - answer_synthesizer — AnswerSynthesizerNode — packages/nl2sql/src/nl2sql/pipeline/nodes/answer_synthesizer/node.py — produce final answer.

Mermaid diagram (main pipeline only):

flowchart TD
    curated_matcher -.->|verified| layer_router
    curated_matcher -.->|generated| datasource_resolver
    datasource_resolver -->|continue| decomposer
    datasource_resolver -->|end| END
    decomposer --> layer_router
    layer_router -->|subgraph| subgraph_exec
    subgraph_exec --> layer_router
    layer_router -->|aggregator| aggregator
    layer_router -->|end| END
    aggregator -->|ok| answer_synthesizer --> END
    aggregator -->|failed| END

The dotted edges exist only when curated answers are configured.

State Contract

State model: GraphState (packages/nl2sql/src/nl2sql/pipeline/state.py).

Field ownership, reducers, and lifecycle are defined in ../graph_state.md.

Step-by-Step Execution Flow

With curated answers configured, curated_matcher runs first. On a verified hit it has already run the query and filled decomposer_response, execution_dag, artifact_refs and subgraph_outputs as for one sub-query, so layer_router finds the scan done and goes straight to step 7. Any fallback continues at step 1 with nothing in the state changed.
datasource_resolver resolves candidate datasources (vector search only when more than one datasource is registered), filters them by RBAC, then asks the LLM whether any allowed datasource can answer the question.
resolver_route decides:
continue if allowed datasources exist and the question was judged answerable.
end if none exist, the question is not answerable, or the response is missing.
decomposer uses the LLM to produce SubQuery objects and combine groups, then builds a deterministic ExecutionDAG from them in code. The DAG is a pure function of that output, so it is not a node of its own; a DAG that cannot be built is reported as PLANNER_FAILED and the run ends at the router.
layer_router inspects the DAG and current artifact_refs:
If no DAG or layers, returns END.
If next scan layer is empty, routes to aggregator -- or to END when no scan produced an artifact or execute is False.
For each scan node, resolves a compatible subgraph and sends build_scan_payload.
Each subgraph execution returns artifact_refs, subgraph_outputs, and errors to GraphState.
layer_router is re-entered until all scan-layer nodes produce artifacts.
aggregator executes the DAG using the stored artifact_refs.
aggregator_route ends the run if aggregation failed; otherwise answer_synthesizer summarizes aggregated results into a final answer.
Graph reaches END.

Determinism Guarantees

Determinism guarantees and non-determinism sources are documented in ../determinism.md.

Error Propagation

Failure domains, retry scope, and recovery limitations are centralized in ../failure_recovery.md.

Retry + Recovery

See ../failure_recovery.md for retry scope and recovery behavior.

Performance Characteristics

Blocking calls include LLM requests in datasource_resolver, decomposer and answer_synthesizer.
The decomposer's DAG construction and the aggregator are CPU-bound.
Subgraph executions are dispatched per scan layer and can run in parallel via LangGraph routing.
Overall pipeline is executed on a single-worker thread pool created per run and guarded by settings.pipeline.timeout_s.

Observability

Each node uses nl2sql.common.logger.get_logger() with node-specific names.
GraphState.trace_id is propagated into subgraph IDs and executor requests, but no code binds it into logging context (trace_context is defined but not used).
run_with_graph() accepts LangGraph callbacks via config={"callbacks": callbacks}.
No explicit metrics or tracing spans are emitted in the graph code.

Configuration

settings.pipeline.timeout_s (default 60) controls total pipeline timeout. It is read from the runtime (ctx.settings), as is every setting the nodes use.
Subgraph selection is governed by datasource capabilities in the subgraph registry.

Extension Points

Add new subgraphs by constructing them in build_graph() and extending resolve_subgraph() with their required capabilities.
Insert additional nodes in build_graph() by adding nodes/edges to the StateGraph.
Customize routing by replacing build_scan_layer_router() or resolver_route().

Known Limitations

execute=False in build_graph() makes layer_router return END instead of dispatching the aggregator, and is passed to the SQL agent subgraph.
Errors accumulated in GraphState.errors do not alter routing unless explicit routes inspect them.
No trace/span propagation into logging context, despite trace context helpers existing.
Merge reducers for artifact_refs/subgraph_outputs overwrite on key conflicts without ordering guarantees.

Subgraph definition: packages/nl2sql/src/nl2sql/pipeline/graph.py
Routing utilities: packages/nl2sql/src/nl2sql/pipeline/routes.py, packages/nl2sql/src/nl2sql/pipeline/graph_utils.py
State model: packages/nl2sql/src/nl2sql/pipeline/state.py
Node implementations:
packages/nl2sql/src/nl2sql/pipeline/nodes/datasource_resolver/node.py
packages/nl2sql/src/nl2sql/pipeline/nodes/decomposer/node.py
packages/nl2sql/src/nl2sql/pipeline/nodes/decomposer/dag.py
packages/nl2sql/src/nl2sql/pipeline/nodes/aggregator/node.py
packages/nl2sql/src/nl2sql/pipeline/nodes/answer_synthesizer/node.py