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.

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.

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:

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.

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:

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 objects are named dependencies, never embedded in YAML:

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

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.

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:

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.

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]

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:

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

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:

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

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:

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

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

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

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

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:

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

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.

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.