permchain

Get started

pip install permchain

Usage

from permchain import Pregel, channels

grow_value = (
    Pregel.subscribe_to("value")
    | (lambda x: x + x)
    | Pregel.write_to(value=lambda x: x if len(x) < 10 else None)
)

app = Pregel(
    chains={"grow_value": grow_value},
    channels={"value": channels.LastValue(str)},
    input="value",
    output="value",
)

assert app.invoke("a") == "aaaaaaaa"

Check examples for more examples.

Near-term Roadmap

• Iterate on API
  • do we want api to receive output from multiple channels in invoke()
  • do we want api to send input to multiple channels in invoke()
  • Finish updating tests to new API
• Implement input_schema and output_schema in Pregel
• More tests
  • Test different input and output types (str, str sequence)
  • Add tests for Stream, UniqueInbox
  • Add tests for subscribe_to_each().join()
• Add optional debug logging
• Implement checkpointing
  • Save checkpoints at end of each step
  • Load checkpoint at start of invocation
  • API to specify storage backend and save key
• Add more examples
  • human in the loop
  • combine documents
  • agent executor
  • run over dataset
• Fault tolerance
  • Retry individual processes in a step
  • Retry entire step?
• Pregel.stream_log to contain additional keys specific to Pregel
  • tasks: inputs of each chain in each step, keyed by {name}:{step}
  • task_results: same as above but outputs
  • channels: channel values at end of each step, keyed by {name}:{step}