Quick Start

This guide gets you from zero to a running LLM pipeline in under 5 minutes.

1. Install

pip install neograph

NeoGraph depends on langgraph, langchain-core, pydantic, and structlog. These are installed automatically.

You will also need an LLM provider. This guide uses OpenAI, but any LangChain-compatible chat model works (Anthropic, Google, OpenRouter, local models via Ollama, etc.):

pip install langchain-openai

2. Configure the LLM

NeoGraph does not own LLM configuration. You provide two functions: an LLM factory that creates chat model instances, and a prompt compiler that builds message lists for each node. You pass them to compile() (in step 5) as keyword arguments.

import os
from langchain_openai import ChatOpenAI

# Map model tiers to actual models
MODELS = {
    "fast": "gpt-4o-mini",
    "reason": "gpt-4o",
}

llm_factory = lambda tier: ChatOpenAI(
    model=MODELS.get(tier, "gpt-4o-mini"),
    api_key=os.environ["OPENAI_API_KEY"],
)

prompt_compiler = lambda template, data, **kwargs: [
    {"role": "user", "content": f"[{template}] {data}"}
]

The llm_factory receives a tier string (like "fast" or "reason") and returns a BaseChatModel. The prompt_compiler receives a template name and the input data from the previous node, and returns a list of messages.

In production, you would replace the prompt compiler with something that loads real prompt templates and injects context. For now, this inline version works.

3. Define Your Schemas

Every node declares its output type as a Pydantic model. This is how NeoGraph knows what data flows between nodes.

from pydantic import BaseModel

class Claims(BaseModel):
    items: list[str]

class ClassifiedClaims(BaseModel):
    classified: list[dict[str, str]]

class Summary(BaseModel):
    text: str

4. Build the Pipeline

Decorate functions with @node. The parameter name IS the dependency — classify(decompose: Claims) wires classify after the decompose node automatically.

import sys
from neograph import node, construct_from_module

@node(outputs=Claims, prompt='decompose', model='fast')
def decompose() -> Claims: ...

@node(outputs=ClassifiedClaims, prompt='classify', model='fast')
def classify(decompose: Claims) -> ClassifiedClaims: ...

@node(outputs=Summary, prompt='summarize', model='fast')
def summarize(classify: ClassifiedClaims) -> Summary: ...

pipeline = construct_from_module(sys.modules[__name__])

Each @node specifies:

outputs — Pydantic model defining the typed output contract
prompt — template name passed to your prompt_compiler
model — tier string passed to your llm_factory

Mode is inferred: prompt= + model= means LLM call (think mode). Neither means the function body runs as scripted Python.

construct_from_module scans the module for all @node-decorated functions, infers the dependency graph from parameter names, and builds a Construct.

5. Compile and Run

from neograph import compile, run

graph = compile(
    pipeline,
    llm_factory=llm_factory,
    prompt_compiler=prompt_compiler,
)
result = run(graph, input={"topic": "microservice authentication"})

print(result["summarize"].text)

compile() converts the Construct into a LangGraph StateGraph, wires all edges, infers the state schema, and compiles. Your llm_factory and prompt_compiler are passed here as keyword arguments — they are captured per-compile, not registered globally. run() executes the graph and returns the final state with framework internals stripped out.

Every field you pass in input= also flows into config["configurable"], so your prompt compiler can read them via config["configurable"]["topic"]. This is how pipeline-level metadata (topic, project_root, whatever you need) reaches every node without threading it through state manually.

The result is a dict keyed by node name. Each value is the typed Pydantic model that node produced.

Complete Example

Here is the full script, copy-paste-runnable:

import os
import sys
from langchain_openai import ChatOpenAI
from pydantic import BaseModel
from neograph import node, construct_from_module, compile, run

# --- Schemas ---
class Claims(BaseModel):
    items: list[str]

class ClassifiedClaims(BaseModel):
    classified: list[dict[str, str]]

class Summary(BaseModel):
    text: str

# --- LLM Configuration ---
llm_factory = lambda tier: ChatOpenAI(
    model="gpt-4o-mini",
    api_key=os.environ["OPENAI_API_KEY"],
)

prompt_compiler = lambda template, data, **kwargs: [
    {"role": "user", "content": (
        f"Break this topic into 3-5 factual claims: "
        f"{kwargs.get('config', {}).get('configurable', {}).get('topic', 'AI')}"
        if template == "decompose"
        else f"Classify each claim as security/reliability/performance: {data}"
        if template == "classify"
        else f"Summarize these classified claims in one paragraph: {data}"
    )}
]

# --- Pipeline ---
@node(outputs=Claims, prompt='decompose', model='fast')
def decompose() -> Claims: ...

@node(outputs=ClassifiedClaims, prompt='classify', model='fast')
def classify(decompose: Claims) -> ClassifiedClaims: ...

@node(outputs=Summary, prompt='summarize', model='fast')
def summarize(classify: ClassifiedClaims) -> Summary: ...

pipeline = construct_from_module(sys.modules[__name__])

# --- Run ---
graph = compile(
    pipeline,
    llm_factory=llm_factory,
    prompt_compiler=prompt_compiler,
)
result = run(graph, input={"topic": "microservice authentication"})

print(result["summarize"].text)

What Just Happened

Behind the scenes, NeoGraph did the following:

@node decorated each function and created a Node spec with the inferred mode, output type, and upstream dependencies
construct_from_module scanned the module for all @node functions, built the dependency graph from parameter names (classify depends on decompose because it has a parameter named decompose), topologically sorted the nodes, and created a Construct
compile() inferred a Pydantic state model with fields for each node’s output type, created LangGraph node functions that call your llm_factory and prompt_compiler, wired START -> decompose -> classify -> summarize -> END, and compiled the LangGraph StateGraph

You got a compiled graph wrapping a standard LangGraph graph. You can stream it (sync stream() or async astream()), checkpoint it, visualize it, or deploy the underlying graph (via .graph) with LangGraph Platform — NeoGraph does not change the runtime, only the authoring.

Running async

run() has an async twin, arun(), with the identical signature. Use it when your nodes, tools, or llm_factory are async, or to run the pipeline under an existing event loop:

from neograph import arun

result = await arun(graph, input={"topic": "microservice authentication"})
print(result["summarize"].text)

The compiled graph also delegates async facade verbs — aget_state / aget_state_history for inspecting checkpointed state and astream_events for event-level observability. See the API reference for the full delegated set.

Streaming

Instead of waiting for the final state, stream the graph as it runs. NeoGraph exposes top-level stream() (sync) and astream() (async) verbs that mirror run()/arun() — same new-input / resume / crash-recovery modes — and add a stream_mode argument (a str or a list, passed straight to LangGraph):

from neograph import stream

for chunk in stream(graph, input={"topic": "microservice authentication"},
                    stream_mode="updates"):
    print(chunk)  # per-node deltas as each node finishes

The async twin drives the same graph on the event loop — the production surface for SSE / AG-UI consumers:

from neograph import astream

async for chunk in astream(graph, input={"topic": "..."}, stream_mode="values"):
    print(chunk)  # full state snapshot after each step

values and updates chunks are stripped of framework-internal (neo_*) fields; custom, messages, and debug payloads pass through untouched.

Typed progress events

A node body can push a typed domain event onto the stream with emit_progress. Pass a Pydantic model; it is wrapped in a stable envelope ({"neograph_event": "progress", "event_type": "<ClassName>", "data": {...}}) and delivered on the custom stream:

from pydantic import BaseModel
from neograph import node, emit_progress

class Scanned(BaseModel):
    count: int

@node(outputs=Claims)
def decompose(topic: RawText) -> Claims:
    emit_progress(Scanned(count=3))
    return Claims(items=[...])

Consume it by driving the graph with stream/astream and including "custom" in stream_mode. Under a non-streaming driver (run/arun, or a stream without custom) the event has no consumer, so emit_progress warns once rather than dropping it silently.

Next Steps

Add tools to a node with mode="agent" and Tool("name", budget=5) — see the Walkthrough
Fan out with ensemble_n=3 or map_over='field.path' in @node kwargs — see Core Concepts
Add human-in-the-loop with interrupt_when= — see the Walkthrough
Use ForwardConstruct for branching pipelines — see What is NeoGraph?