tketool.pipeline

Typed node and edge orchestration backed by LangGraph. The public API uses Pydantic models and does not expose LangGraph state objects.

pip install tketool.pipeline

Basic graph

from pydantic import BaseModel
from tketool.pipeline import END, START, DirectEdge, Pipeline, node


class Question(BaseModel):
    text: str


class Answer(BaseModel):
    text: str


@node(Question, Answer, node_id="answer")
def answer(value: Question):
    return {"text": value.text.upper()}


graph = Pipeline(Question, Answer).add_node(answer).add_edges(
    DirectEdge(START, "answer"),
    DirectEdge("answer", END),
)

result = graph.invoke({"text": "hello"})

A class node declares the same contract with generics:

from tketool.pipeline import Node


class AnswerNode(Node[Question, Answer]):
    def execute(self, value: Question):
        return {"text": value.text.upper()}

execute() is the only execution method a custom class implements. A graph definition never contains a call field: type selects either a built-in node or an explicitly registered custom node class, and the node owns its behavior.

Explicit data types

Every public node and graph input/output model is checked recursively when it is constructed or resolved from YAML. Any, object, bare BaseModel, bare list/dict, and unresolved generic fields fail immediately with a field path. Use scalar types, named Pydantic models, list[Item], and dict[str, Value]:

class Item(BaseModel):
    name: str
    scores: dict[str, float]


class Batch(BaseModel):
    items: list[Item]
    selected: Item | None = None

validate_model_types(Batch) exposes the same check to editors and tooling. Explicit Union, Optional, Literal, Annotated, TypedDict, recursive models, and standard scalar types such as datetime, UUID, and Decimal remain supported. Runtime resource classes and callable fields are rejected. Models using extra="allow" must declare __pydantic_extra__: dict[str, T] with an explicit value type; undeclared extra data cannot bypass validation. Existing explicitly parameterized tuple, set and mapping annotations also remain supported. Runtime dependencies and execution context are outside this data contract.

For intentionally dynamic JSON boundaries, pydantic.JsonValue explicitly means string, number, boolean, null, recursively typed list, or string-keyed dictionary. The generic split node uses this union for items and the memory nodes use dict[str, JsonValue] for metadata. Opaque Python objects are rejected at these boundaries. Domain-specific inputs should use their actual item or value type. SplitSourceRef is an explicit TypedDict for fragment provenance and retains dictionary access. JSON Schema still uses the standard "type": "object" notation for dictionaries and named records; it is not an unconstrained Python object field.

Per-execution shared context

Use a graph context for resources that change on every invocation, such as the current tenant, request-scoped metadata, permission object, data source, or prompt pool. The caller defines the context class; the pipeline creates one immutable ExecutionContext wrapper for the run and gives that same wrapper to every node, including each for iteration and retry attempts.

from dataclasses import dataclass
from tketool.pipeline import ExecutionContext, Node, Pipeline


@dataclass(frozen=True)
class RequestContext:
    tenant_id: str


class AnswerNode(Node[Question, Answer]):
    def execute(
        self,
        value: Question,
        context: ExecutionContext[RequestContext],
    ):
        session = context.session
        return {"text": f"{session.tenant_id}:{value.text}"}


graph = Pipeline(Question, Answer, context_type=RequestContext).add_node(
    AnswerNode(node_id="answer")
).add_edges(DirectEdge(START, "answer"), DirectEdge("answer", END))
result = graph.invoke(
    {"text": "hello"},
    context=RequestContext(
        tenant_id="tenant-a",
    ),
)

When context_type is declared, invoke, ainvoke, stream, and astream require an instance of that class. Different concurrent invocations receive different Session values. Graph services belong to GraphEnvironment. RunConfig(metadata=...) remains available as ExecutionContext.metadata for run labels and tracing.

The context type can be part of a saved graph definition, but context instances cannot:

types:
  models:
    Question: myapp.models:Question
    Answer: myapp.models:Answer
  contexts:
    RequestContext: myapp.context:RequestContext
  nodes:
    answer: myapp.nodes.answer:AnswerNode

pipeline:
  name: answer_graph
  input: Question
  output: Answer
  context: RequestContext

Session resources travel through LangGraph's run-scoped context, not graph state. They are therefore absent from execution tokens and checkpoints. The caller owns their construction, thread safety, and shutdown. In contrast, dependencies={...} passed to Pipeline.load/save supplies graph-lifetime constructor dependencies such as the built-in PromptNode's fixed pool.

YAML graph definitions

Definitions are versioned, strict, data-only YAML. types gives short names to importable Python symbols; pipeline.nodes creates node instances; flow owns all connections.

version: 3

types:
  models:
    Question: myapp.models:Question
    Answer: myapp.models:Answer
  nodes:
    answer: myapp.nodes.answer:AnswerNode

pipeline:
  name: answer_graph
  input: Question
  output: Answer

  nodes:
    answer:
      type: answer
      params:
        prefix: "result: "
      policy:
        attempts: 3
        retry_on: [ConnectionError, TimeoutError]
        timeout_seconds: 20
        metadata:
          role: llm

  flow:
    start: answer
    steps: []
    end: answer

Load only application modules that the caller explicitly trusts:

from tketool.pipeline import Pipeline

graph = Pipeline.load(
    "answer_graph.yaml",
    allowed_modules=["myapp"],
)
result = graph.invoke({"text": "hello"})

allowed_modules=["myapp"] allows myapp and its submodules. Imports outside that list fail with ComponentResolutionError. YAML uses safe_load; Python object tags and unknown fields are rejected.

Type declarations

Built-in retry names are Exception, RuntimeError, ValueError, TypeError, ConnectionError, and TimeoutError. Other errors must be declared in types.errors. Paths use module:Symbol; filesystem paths are not accepted.

Built-in node types

Runtime services belong to a graph environment and never appear in builtin params:

from tketool.pipeline import GraphEnvironment, Pipeline

environment = GraphEnvironment({"prompt_pool": prompt_pool, "memory": memory})
graph = Pipeline.load("graph.yaml", allowed_modules=["myapp"], environment=environment)
# Or Pipeline(Input, Output, environment=environment) for programmatic graphs.

Required resource keys are prompt_pool for Prompt and memory for both memory nodes. Compilation checks their Protocols. Bindings are immutable; objects and lifecycle stay caller-owned. Each execution receives the graph's environment through ExecutionContext, outside checkpoints and serialized definitions.

Split and memory nodes

Split exposes three settings: rule (auto, text, collection), text and collection. Text options are chunk_size=1000, chunk_overlap=0; overlap must be smaller than size. Collection options are max_items=1000, max_chars=None, text_field=None, context_items=0, id_field=None.

nodes:
  split:
    type: split
    params:
      rule: auto
      text: {chunk_size: 800, chunk_overlap: 80}
      collection:
        max_items: 20
        max_chars: 1500
        text_field: text
        context_items: 2
        id_field: id

auto selects by input type; explicit modes reject mismatches. Text preserves separators and whitespace using the built-in Chinese/English separator order. Collection limits apply only to new records. Historical context is a contiguous suffix bounded by context_items and may increase total window characters beyond max_chars. Oversized single records fail clearly. items contains windows and source_refs preserves provenance; output shape is fixed. Custom splitting strategies use custom nodes rather than injecting a hidden global override. Old flat split parameters and behavior switches are rejected.

from tketool.pipeline import ExecutionContext, GraphEnvironment, MemoryNode, MemoryRecallNode

context = ExecutionContext(run_id="example", environment=GraphEnvironment({"memory": memory}))
saved = MemoryNode(kind="preference", tags=["profile"]).invoke(
    {"content": "用户喜欢乌龙茶", "idempotency_key": "request-1"}, context=context
)
recalled = MemoryRecallNode(limit=3, kinds=["preference"]).invoke(
    {"query": "用户喜欢喝什么？"}, context=context
)

Node-level kind, tags, metadata, idempotency_key, limit, kinds, tags, and using are defaults. Non-null values in each invocation override them; invocation metadata is merged over configured metadata. Semantic or entity recall still requires the corresponding feature to be configured on create_memory.

nodes:
  remember:
    type: memory
    params:
      kind: preference
      tags: [profile]
  recall:
    type: memory_recall
    params:
      limit: 3
      kinds: [preference]
      using: [lexical]

graph = Pipeline.load(
    "graph.yaml",
    allowed_modules=["myapp"],
    environment=GraphEnvironment({"memory": memory}),
)

Flow and edge definitions

The default registry includes four edge types. Version 3 YAML uses params for configuration, an explicit registered type, and one from/to pair per step. All paths are declarative: . selects the entire output, .field.nested selects fields, and .items.0 selects a list element. Missing paths and invalid comparisons fail explicitly; paths never execute code.

Keyword properties follow source and target; YAML conditional steps must explicitly declare operator. Operators are eq, ne, gt, ge, lt, le, in, not_in, contains, not_contains, truthy, falsy, is_none, and not_none. For in, the selected value must be in the configured value; for contains, the selected container must contain the configured value.

- type: direct
  from: prepare
  to: inspect
- type: selector
  from: inspect
  to: answer
  params: {selector: .result.text, into: text}
- type: conditional
  from: answer
  to: publish
  params: {selector: .confidence, operator: ge, value: 0.8}

These are separate connection examples, not a complete graph. A source may have multiple independent condition edges; they do not implement a first-match switch. Every invocation must produce exactly one final output: zero or multiple active finish paths fail, including multiple active edges into END from the same source.

Node inputs and activation

A single edge delivers its output directly. With activation="all" (default), multiple incoming edges produce a tuple in edge declaration order, independent of upstream execution order. Declare Node[tuple[Left, Right], Answer]; execute receives the two validated model instances as a Python tuple. Duplicate field names remain independent. Missing conditional positions fail validation; values are never shifted or filled silently. Loop invocations remain isolated.

activation="each" retains one invocation per delivery and a single-value input contract. This does not enable parallel scheduling or exactly-once side effects. For explicit ports, YAML can declare types.models.Pair: {tuple: [Left, Right]}. Fixed custom node types derive their tuple contract directly from Python.

flow.start creates the single direct entry edge. flow.end declares one or more possible finish nodes. A non-direct finish edge is written explicitly with to: __pipeline_end__ and its source also listed in flow.end. Put input selection after an ordinary entry node; START accepts only a direct edge.

Structured for, end, break, and continue

A ForEdge must name exactly one ForEndNode. Its body may contain multiple nodes and nested loops. The selected value must be a list. item_key / YAML as optionally wraps each element for a named node input field.

ForEndNode(Model, node_id="done", into=None) collects one final value for each normal iteration. Model is a RootModel[list[T]], or a model with the list field named by into. Other fields must have defaults. A body's edge into its paired end carries one item, while the end node itself receives and emits the complete list. Input order is preserved and nested lists are not flattened. An empty list executes the end with [] without running the body.

ContinueNode(ItemModel, node_id="skip", end="done") excludes the current item and advances. BreakNode(ItemModel, node_id="stop", end="done") excludes the current item and ends that loop before later items can start. These control nodes have no outgoing edges and must target the innermost enclosing loop's end. Use condition edges to choose a control node or a normal body path.

If no condition edge in a loop body matches, the current item is skipped. More than one active outgoing body edge is an error; body branch conditions must be mutually exclusive. A failure stops the invocation instead of silently dropping the item. This adapter executes serially and rejects max_concurrency > 1, including for iterations.

This complete example doubles a root list without creating a wrapper field:

from pydantic import RootModel
from tketool.pipeline import END, START, DirectEdge, ForEdge, ForEndNode, Pipeline, node


class Number(RootModel[int]):
    pass


class Numbers(RootModel[list[int]]):
    pass


@node(Numbers, Numbers, node_id="source")
def source(value):
    return value


@node(Number, Number, node_id="double")
def double(value):
    return value.root * 2


graph = Pipeline(Numbers, Numbers).add_nodes(
    source, double, ForEndNode(Numbers, node_id="done")
).add_edges(
    DirectEdge(START, "source"),
    ForEdge("source", "double", end="done"),
    DirectEdge("double", "done"),
    DirectEdge("done", END),
)
assert graph.invoke([3, 1, 2]).root == [6, 2, 4]
assert graph.invoke([]).root == []

This complete YAML example needs only built-in nodes and models. It splits a collection into single-item windows, traverses those windows and returns them:

version: 3
types:
  models:
    Input: tketool.pipeline:SplitInput
    Output: tketool.pipeline:SplitOutput
pipeline:
  name: window_iteration
  input: Input
  output: Output
  nodes:
    split:
      type: split
      params: {rule: collection, collection: {max_items: 1}}
    process:
      type: identity
      input: Input
    done:
      type: for_end
      input: Output
      params: {into: items}
  flow:
    start: split
    steps:
      - type: for
        from: split
        to: process
        params: {selector: .items, item_key: value, end: done}
      - type: selector
        from: process
        to: done
        params: {selector: .value}
    end: done

Save it as windows.yaml, then run Pipeline.load("windows.yaml", allowed_modules=["tketool.pipeline"]).invoke({"value": [1, 2]}); the output has items == [[1], [2]]. Replace the identity node with a typed window-processing node to operate on each window.

Versions 1 and 2 are rejected. Rewrite edge parameters under params; rename as to item_key. Custom Edge[Input, Output] implementations register using ComponentRegistry.register_edge. types.edges can reference importable Edge classes; ordinary callable edge aliases are not supported.

Project component registry

from tketool.pipeline import ComponentConfig, default_registry

class AnswerConfig(ComponentConfig):
    prefix: str = "result: "

registry = default_registry()
registry.register_node("answer", AnswerNode, config_model=AnswerConfig)

Export this registry from your module and add registry: myapp.components:registry to the GraphDefinition. The config model, Python generic ports, factory and optional pure definition_validator live in Pipeline's registration. Studio reads this same contract for its palette, configuration fields, type display and validation; adding a component requires no Studio configuration or code.

Use describe_registry / inspect_definition from tketool.pipeline or the JSON stdin CLI python -m tketool.pipeline. Inspection never constructs business nodes; the allowlisted module itself is trusted Python and may run import code. A loaded definition retains constructor data for save/reload, so change that definition and rebuild instead of mutating a loaded instance's configuration. See examples/pipeline/registered_agent in the repository for a runnable custom node and edge example.

Custom node rules

A reusable custom node is a module-level class. Declare input/output through the generic base and declare parameter fields with ConfigBase subclasses:

from tketool.pipeline import ComponentConfig, Node, StringConfig, IntegerConfig


class AnswerNode(Node[Question, Answer]):
    class Config(ComponentConfig):
        prefix: str = StringConfig(default="result: ", describe="Answer prefix", icon="type")
        repeats: int = IntegerConfig(default=1, ge=1, le=3, describe="Repeat count")

    def execute(self, value: Question):
        params = self.parameters
        return Answer(text=params.prefix + value.text.upper() * params.repeats)

The base constructor accepts AnswerNode(prefix="OK: ", repeats=2) or AnswerNode(parameters=AnswerNode.Config(...)). Registration discovers Config. Node and Edge share automatic validation, parameter assembly and YAML export; Edge implements prepare() and also reads self.parameters.

ConfigBase defines each field's type, candidates, describe, icon, default and constraints. Its subclasses are StringConfig, IntegerConfig, FloatConfig, BooleanConfig, SelectConfig, ListConfig, ObjectConfig, DependencyConfig and TypedConfig. SelectConfig(candidates=[...]) validates choices in Python as well as Studio. Python type annotations determine value types; descriptors stay in the field definitions. ComponentConfig.config_fields() exposes normalized field descriptors, including inherited fields and ordinary Pydantic declarations. model_json_schema() emits description/enum/x-icon/x-config-type and constraints; Studio renders the same schema. Nested ComponentConfig objects retain metadata.

Configuration validates defaults, forbids unknown fields and freezes attribute assignment. Nested collections are revalidated before execution, even on a cached compiled graph. Runtime dependencies are injected after placeholder validation and preserve object identity; save through Pipeline.save/to_definition with named dependencies instead of dumping live resource objects directly.

Existing explicit constructors remain supported. Automatic export reads the configuration model's matching instance attributes; absent reconstruction values fail explicitly. Override definition_params() only for custom storage layouts. Exported values must be YAML data or named dependency objects. Replace/rebuild a loaded definition to change its configuration. See the runnable config_fields example.

The decorator form remains supported:

from tketool.pipeline import node


@node(Question, Answer, node_id="answer")
def answer(value: Question):
    return Answer(text=value.text.upper())

Reference the exported decorated symbol in types.nodes. Decorated nodes and plain functions cannot receive YAML params; use a class when configuration is required. Put custom transformations and predicates in typed processing nodes; edges only select data or use the supported declarative comparisons. Local classes and lambdas cannot be serialized as importable node symbols.

Generate, save, and reload

definition = graph.to_definition()
yaml_text = graph.to_yaml()
graph.save("graph.yaml")

same_graph = Pipeline.from_definition(
    definition,
    allowed_modules=["myapp"],
    environment=GraphEnvironment({"prompt_pool": prompt_pool}),
)
same_graph = Pipeline.from_yaml(
    yaml_text,
    allowed_modules=["myapp"],
    environment=GraphEnvironment({"prompt_pool": prompt_pool}),
)
same_graph = Pipeline.load(
    "graph.yaml",
    allowed_modules=["myapp"],
    environment=GraphEnvironment({"prompt_pool": prompt_pool}),
)

These methods save the graph definition, not execution state. LangGraph checkpointers remain the separate mechanism for saving a running thread's state and resuming it.

Prompt node

PromptNode(Question, Answer, prompt_key="answer_question") exposes only the prompt key. The graph environment supplies prompt_pool; calls are always serialized and output is exactly the validated model result. There are no preserve_input, serialize_calls or result_adapter switches. Use typed nodes and edges to compose values explicitly. For tuple input, the prompt adapter supplies a positional inputs template value without merging its elements' fields.

Execution, events, and limits

• invoke/ainvoke return the graph output model. RootModel supports scalar and list graph/node boundaries.
• stream/astream emit runtime-neutral NodeEvent values and validate the final output when the stream is exhausted. Consume the entire stream to see completion failures; receiving one node event does not prove graph success.
• For body events carry the current zero-based iteration index. A nested loop body reports its inner index; its completed end reports the enclosing outer index. The outermost end has no item index.
• RunConfig(recursion_limit=10000) is the default LangGraph step budget. Sequential loops consume steps for each executed node; increase this explicit limit for larger workloads. This adapter rejects max_concurrency > 1.
• NodeConfig(timeout=...) is enforced by ainvoke/astream and raises the public NodeTimeoutError. Timed synchronous work may continue in its worker thread after the async graph stops waiting. Sync entrypoints do not support node timeouts. A synchronous node with a timeout and timeout-matching retries is rejected at compile time because an abandoned thread cannot be cancelled. Native async nodes await cancellation cleanup before a timeout retry.
• Native async def execute() is supported through ainvoke/astream; synchronous entrypoints reject async nodes explicitly.
• compiled.resume(config=RunConfig(configurable={"thread_id": "run-1"}), context=...) and await compiled.aresume(...) continue an unfinished checkpoint. Compile with a caller-owned checkpointer and reuse its thread ID. Recovery creates a fresh Session, requires fresh resource bindings, and preserves the original run ID. Missing/completed checkpoints are rejected. Persistent checkpointers support reopening and recompiling; this does not guarantee exactly-once external effects.

The pre-2.0 stage/decorator API (invoke_all, invoke_one, ExecutionBoard) is intentionally removed rather than emulated with different semantics.

Explicit data composition

Prompt returns its model result and ForEnd returns its collected results. To use an earlier value together with a later result, connect both through edges into a typed tuple-input node. Data composition is visible in the graph rather than controlled by preservation flags.

Node input bindings

A node can select individual input fields from its incoming value or from the run's Session context. Bindings run before the node input model is validated and explicitly bound fields override the corresponding incoming fields:

pipeline:
  context: SessionData
  nodes:
    report:
      type: report
      input: ReportInput
      output: Report
      bindings:
        text: {source: input, selector: .text}
        audience: {source: session, selector: .audience}

Declare SessionData in types.contexts, then call pipeline.invoke(value, context=SessionData(...)). Existing selector syntax is shared with edges. Bindings require object inputs and select JSON-compatible business data, including nested Pydantic models. Runtime resource objects cannot be bound. The selected Session values are captured when each run starts and copied for every consumer; bindings do not add node activations. Legacy context-aware code still receives the original ExecutionContext.session object.

Optional component display metadata

ComponentMetadata is a runtime-neutral optional hint for any visualizer or documentation tool. All fields default to the empty string; Pipeline execution never reads them. It does not import Studio or an icon library.

from pydantic import BaseModel
from tketool.pipeline import ComponentMetadata, IdentityNode

class Message(BaseModel):
    text: str

step = IdentityNode(
    Message,
    node_id="forward",
    display=ComponentMetadata(
        title="传递消息",
        description="将已验证的消息交给后续节点",
        icon="arrow-right",
    ),
)

Node and Edge constructors accept display. Component registration retains title and description and also accepts icon. Non-empty instance fields override registration values, which override class-level ComponentMetadata defaults. An empty field inherits; an unnamed catalog entry falls back to its component ID. Unknown icon strings are valid and consumers decide their fallback rendering.

In Graph YAML, display: {title: ..., description: ..., icon: ...} belongs beside type on a node or edge, not inside execution params or policy.metadata. Load/save preserves these values and registry inspection returns the resolved metadata. Existing Graph definitions without display remain valid.