Metadata-Version: 2.4
Name: tketool.pipeline
Version: 2.0.1
Summary: Typed node and edge pipelines powered by LangGraph
Author-email: Ke <jiangke1207@icloud.com>
License-Expression: MIT
Project-URL: Homepage, https://pypi.org/project/tketool.pipeline/
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Operating System :: OS Independent
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: tketool.core==2.0.1
Requires-Dist: langchain-core<2,>=1.4.7
Requires-Dist: langchain-text-splitters<2,>=1.1.2
Requires-Dist: langgraph<1.3,>=1.2.11
Requires-Dist: langgraph-checkpoint<5,>=4.2
Requires-Dist: pydantic<3,>=2.10
Requires-Dist: PyYAML<7,>=6.0.3
Provides-Extra: llm
Requires-Dist: tketool.llm==2.0.1; extra == "llm"
Provides-Extra: test
Requires-Dist: pytest<9,>=8; extra == "test"

# tketool.pipeline

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

```bash
pip install tketool.pipeline
```

## Basic graph

```python
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:

```python
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]`:

```python
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.

```python
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:

```yaml
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.

```yaml
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:

```python
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

| Section | Value | Purpose |
| --- | --- | --- |
| `types.models` | `package.module:ModelClass` | Pydantic input/output models used by the graph or built-in nodes |
| `types.contexts` | `package.module:ContextClass` | optional caller-defined per-execution context type |
| `types.nodes` | `package.module:NodeClass` | custom `Node` subclass or decorated node |
| `types.errors` | `package.module:ErrorClass` | custom exception classes allowed in `policy.retry_on` |

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

| `type` | Required definition | Behavior |
| --- | --- | --- |
| `for_end` | `input` aggregate model, optional `params.into` | emits the paired loop's result list once |
| `break` | `input` current-item model, `params.end` | stops the innermost loop, excluding the current item |
| `continue` | `input` current-item model, `params.end` | skips the current item in the innermost loop |
| `identity` | `input`, optional `output` | validates and forwards the input |
| `prompt` | `input`, `output`, `params.prompt_key` | gets an invoker from an injected prompt pool and validates its result |
| `split` | optional rule/window params | splits `SplitInput.value` into ordered `SplitOutput.items` |
| `memory` | optional write defaults | stores `MemoryInput.content` and returns `MemoryOutput.memory` |
| `memory_recall` | optional recall defaults | recalls ranked records for `MemoryRecallInput.query` |

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

```python
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`.

```yaml
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.

```python
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`.

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

```python
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.

| Python API | YAML `type` | Properties and behavior |
| --- | --- | --- |
| `DirectEdge(source, target)` | `direct` | passes the complete output |
| `SelectorEdge(source, target, selector=".", into=None)` | `selector` | selects a value; optional `into` wraps it as `{field: value}` |
| `ForEdge(source, target, end=..., selector=".", item_key=None)` | `for` | selects a list, executes the body sequentially; YAML `params.item_key` is Python `item_key` |
| `ConditionEdge(source, target, selector=".", operator="eq", value=None)` | `conditional` | tests the selected value and passes the complete output only when true |

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`.

```yaml
- 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:

```python
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:

```yaml
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

```python
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:

```python
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](../../examples/pipeline/config_fields/README.md).

The decorator form remains supported:

```python
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

```python
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:

```yaml
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.

```python
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.
