Non-Node Parameters

Not every parameter in a @node function is an upstream dependency. NeoGraph recognizes four kinds of parameters and resolves each one differently at runtime.

The four kinds

Kind	Annotation	Resolved from	Example
Upstream node	Plain type (`Claims`)	Graph state, by parameter name	`decompose: Claims`
Runtime input	`FromInput[T]`	`run(input={...})`	`topic: FromInput[str]`
Shared resource	`FromConfig[T]`	`config['configurable']`	`limiter: FromConfig[RateLimiter]`
Constant	Default value	Compile-time default	`max_items: int = 10`

The framework classifies parameters at decoration time (for FromInput and FromConfig) and at construct_from_module time (for constants — because it needs the full set of decorated names to distinguish “unknown upstream” from “has a default”).

FromInput[T]

Values injected from run(input={...}). Use this for data that varies per execution — user queries, document IDs, configuration overrides.

from neograph import node, FromInput

@node(output=RawText)
def fetch(doc_id: FromInput[str]) -> RawText:
    # doc_id comes from run(graph, input={"doc_id": "DOC-42"})
    return RawText(text=f"Contents of {doc_id}")

At runtime, run() injects all input fields into config["configurable"]. The framework reads config["configurable"]["doc_id"] and passes it to the function. If the key is absent, None is passed.

FromConfig[T]

Shared resources injected via config['configurable']. Use this for objects that live across the entire pipeline — rate limiters, database connections, API clients.

from neograph import node, FromConfig

class RateLimiter:
    def check(self) -> bool:
        return True

@node(output=Result)
def process(data: UpstreamData, limiter: FromConfig[RateLimiter]) -> Result:
    if not limiter.check():
        return Result(status="rate_limited")
    return Result(status="ok")

Pass the resource when running:

graph = compile(pipeline)
result = run(graph, input={"node_id": "x"}, config={
    "configurable": {"limiter": RateLimiter()}
})

Both FromInput and FromConfig resolve from config["configurable"] at runtime. The distinction is semantic — FromInput communicates “per-execution data” while FromConfig communicates “shared infrastructure.”

Default values

Parameters with a default value that don’t match any upstream @node are treated as compile-time constants.

@node(output=Summary)
def summarize(claims: Claims, max_items: int = 10, verbose: bool = False) -> Summary:
    items = claims.items[:max_items]
    if verbose:
        return Summary(text=f"Processed {len(items)} of {len(claims.items)}")
    return Summary(text=f"{len(items)} items")

The default is captured at decoration time and passed to every invocation. Constants are not part of the graph topology — they don’t create edges.

Mixed parameters

A single function can use all four kinds:

from neograph import node, FromInput, FromConfig
from pydantic import BaseModel

class Claims(BaseModel, frozen=True):
    items: list[str]

class Scores(BaseModel, frozen=True):
    ratings: dict[str, float]

class Report(BaseModel, frozen=True):
    summary: str

class RateLimiter:
    def wait(self): pass

@node(output=Report)
def summarize(
    claims: Claims,                        # upstream node — wired by parameter name
    scores: Scores,                        # upstream node — fan-in, second edge
    topic: FromInput[str],                 # from run(input={"topic": "security"})
    rate_limiter: FromConfig[RateLimiter], # from config["configurable"]["rate_limiter"]
    max_items: int = 10,                   # compile-time constant
) -> Report:
    rate_limiter.wait()
    top = sorted(scores.ratings.items(), key=lambda x: -x[1])[:max_items]
    lines = [f"{claim}: {score:.1f}" for claim, score in top]
    return Report(summary=f"Topic: {topic}\n" + "\n".join(lines))

Running the pipeline:

import sys
from neograph import construct_from_module, compile, run

pipeline = construct_from_module(sys.modules[__name__])
graph = compile(pipeline)

result = run(graph, input={
    "node_id": "review-001",
    "topic": "security audit",
}, config={
    "configurable": {"rate_limiter": RateLimiter()}
})

Resolution order

When construct_from_module processes a parameter:

If annotated FromInput[T] — classified as runtime input.
If annotated FromConfig[T] — classified as shared resource.
If the name matches a decorated @node in the module — classified as upstream dependency (creates an edge).
If the parameter has a default value — classified as constant.
Otherwise — ConstructError with a message listing available @node names.

This means a parameter named claims with a default value of [] will be treated as an upstream dependency if there’s a @node function called claims in the module — the upstream match takes priority over the default.