Metadata-Version: 2.4
Name: tketool.pipeline
Version: 1.4.0
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: 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==1.4.0; 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.

## 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 parallel map workers and retry attempts.

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


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


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)
result = graph.invoke(
    {"text": "hello"},
    context=RequestContext(
        tenant_id="tenant-a",
        knowledge_source=knowledge_source,
        prompt_pool=prompt_pool,
    ),
)
```

When `context_type` is declared, `invoke`, `ainvoke`, `stream`, and `astream`
require an instance of that class. Different concurrent invocations receive
different wrappers and resources. `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 `Send` payloads 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: 1

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, decorated node, or module-level function |
| `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 |
| --- | --- | --- |
| `identity` | `input`, optional `output` | validates and forwards the input |
| `prompt` | `input`, `output`, `params.prompt_key`, `params.prompt_pool` | 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` | `params.memory` dependency, optional write defaults | stores `MemoryInput.content` and returns `MemoryOutput.memory` |
| `memory_recall` | `params.memory` dependency, optional recall defaults | recalls ranked records for `MemoryRecallInput.query` |

Runtime objects are named dependencies, never embedded in YAML:

```yaml
nodes:
  answer:
    type: prompt
    input: Question
    output: Answer
    params:
      prompt_key: answer_question
      prompt_pool:
        dependency: prompt_pool
      serialize_calls: true
```

```python
graph = Pipeline.load(
    "graph.yaml",
    allowed_modules=["myapp"],
    dependencies={"prompt_pool": prompt_pool},
)
```

### Split and memory nodes

`SplitNode` has a fixed `value -> items` contract so its output can feed a
`MapEdge`. `rule: auto` selects text or collection behavior from the input;
`text` and `collection` make a type mismatch fail explicitly. `chunk_overlap`
must be smaller than `chunk_size`. Text rules use recursive character splitting.
The default priority is paragraph (`\n\n`), line, Chinese/ASCII sentence
punctuation, comma/colon, whitespace, and finally character-level fallback, so
both paragraph text and long punctuation-delimited strings work without extra
configuration. Collection rules use item counts. Boundary whitespace is
preserved by default; set `strip_whitespace: true` only when trimming chunk
edges is intentional.

```yaml
nodes:
  split:
    type: split
    params:
      rule: text
      chunk_size: 800
      chunk_overlap: 80
      separators: ["\n\n", "\n", "。", "！", "？", "；", ".", "!", "?", ";", "，", ",", "：", ":", "、", " ", ""]
      keep_separator: end
      strip_whitespace: true
```

`separators` is ordered. To split strictly by paragraphs, configure
`["\n\n", ""]`; to prioritize punctuation, configure for example
`["。", "！", "？", ".", "!", "?", ""]`. Keep the final empty string so an
individual overlong paragraph or sentence can still be bounded by
`chunk_size`.

Pass a custom object implementing `split(value) -> list` as a named
`params.splitter` dependency when built-in rules are insufficient.

The two memory nodes use the existing `tketool.llm.memory.Memory` contract.
They do not create a backend, choose a tenant/space, own connection shutdown,
or embed a live memory object in YAML. Build the memory once and inject the same
instance by name:

```python
from tketool.llm.memory import create_memory
from tketool.pipeline import MemoryNode, MemoryRecallNode
from tketool.storage import MemoryBackend

memory = create_memory(MemoryBackend(), space="user-42")

saved = MemoryNode(
    memory=memory,
    kind="preference",
    tags=["profile"],
).invoke({"content": "用户喜欢乌龙茶", "idempotency_key": "request-1"})

recalled = MemoryRecallNode(
    memory=memory,
    limit=3,
    kinds=["preference"],
    tags=["profile"],
    using=["lexical"],
).invoke({"query": "用户喜欢喝什么？"})
```

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:
      memory: {dependency: user_memory}
      kind: preference
      tags: [profile]
  recall:
    type: memory_recall
    params:
      memory: {dependency: user_memory}
      limit: 3
      kinds: [preference]
      using: [lexical]
```

```python
graph = Pipeline.load(
    "graph.yaml",
    allowed_modules=["myapp"],
    dependencies={"user_memory": memory},
)
```

### Flow and edge definitions

The YAML exposes four built-in connection forms. Each `steps` item must match
exactly one form.

Direct connection, including parallel fan-out:

```yaml
- from: prepare
  to: normalize
- from: prepare
  to: [search, summarize]
```

A direct edge may adapt the source output with a declared module-level
callable. The alias is resolved through the same import allowlist as nodes:

```yaml
types:
  edges:
    to_memory: myapp.edges:to_memory

pipeline:
  flow:
    steps:
      - from: split
        to: remember
        transform: to_memory
```

Field-based conditional routing maps a source output field to targets:

```yaml
- from: router
  switch:
    field: route
    cases:
      fast: quick_answer
      deep: researched_answer
```

When the route is derived rather than part of the source model, declare a
module-level condition instead. A switch must contain exactly one of `field`
or `condition`:

```yaml
types:
  edges:
    choose_result: myapp.edges:choose_result

pipeline:
  flow:
    steps:
      - from: answer
        switch:
          condition: choose_result
          cases:
            success: done
            fallback: retry
```

Wait for all listed branches, then bind each complete node output to a field of
the target node input:

```yaml
- wait_for: [search, summarize]
  then: merge
  inputs:
    search_result: search
    summary_result: summarize
```

Use `into` instead of `inputs` when the target expects an ordered list:

```yaml
- wait_for: [a, b]
  then: merge
  into: results
```

Map a list field to parallel worker executions and collect results in original
input order:

```yaml
- foreach: split.items
  run: worker
  as: item
  collect:
    then: merge
    into: results
    preserve_order: true
```

`flow.start` is the single graph entry. `flow.end` is one node id or a list of
possible finish nodes. Version 1 requires an acyclic graph, waits for every
source in a join, preserves map order, and does not serialize a race/first-result
join or a conditional route directly to `END`.

### Custom node rules

A reusable custom node is a module-level class. Declare input/output through
the generic base, put behavior in `execute()`, and accept `node_id`, `config`,
plus any values listed under YAML `params`:

```python
from tketool.pipeline import Node


class AnswerNode(Node[Question, Answer]):
    def __init__(self, node_id=None, *, prefix="", config=None):
        self.prefix = prefix
        super().__init__(node_id=node_id, config=config)

    def execute(self, value: Question):
        return Answer(text=f"{self.prefix}{value.text.upper()}")

    def definition_params(self):
        return {"prefix": self.prefix}
```

`definition_params()` is needed only when a programmatically created custom
node must be saved and reconstructed. Values must be YAML data or objects named
in the `dependencies` mapping passed to `save()`.

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. Importable module-level direct-edge transforms and switch conditions
are declared through `types.edges`. Local classes, lambdas, conditional-edge
transforms, result adapters, and other undeclared runtime callables remain
non-serializable.

### Generate, save, and reload

```python
definition = graph.to_definition(dependencies={"prompt_pool": prompt_pool})
yaml_text = graph.to_yaml(dependencies={"prompt_pool": prompt_pool})
graph.save("graph.yaml", dependencies={"prompt_pool": prompt_pool})

same_graph = Pipeline.from_definition(
    definition,
    allowed_modules=["myapp"],
    dependencies={"prompt_pool": prompt_pool},
)
same_graph = Pipeline.from_yaml(
    yaml_text,
    allowed_modules=["myapp"],
    dependencies={"prompt_pool": prompt_pool},
)
same_graph = Pipeline.load(
    "graph.yaml",
    allowed_modules=["myapp"],
    dependencies={"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` receives the existing prompt pool directly. The pipeline package
only relies on the pool's `get_invoker(key)` structural contract, so it does
not own the model, API key, prompt registry, or pool lifecycle.

```python
from tketool.pipeline import PromptNode

prompt_node = PromptNode(
    Question,
    Answer,
    node_id="answer",
    prompt_key="answer_question",
    prompt_pool=prompt_pool,
)
```

Prompt calls are serialized by default because `PromptInvokerPool` caches
mutable invokers. A caller with a verified thread-safe pool can opt into
same-node parallel calls with `serialize_calls=False`.

## Routing and lists

- `ConditionalEdge` selects a declared route.
- `MapEdge` uses LangGraph `Send` internally to execute list items concurrently.
- `JoinEdge` waits for all branches and restores mapped results to input order.
- `RunConfig(max_concurrency=n)` bounds one invocation's concurrency.
- `invoke`/`ainvoke` return the graph output model; `stream`/`astream` emit
  runtime-neutral `NodeEvent` values.
- `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.

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