permchain

Get started

pip install permchain

Usage

from permchain import InMemoryPubSubConnection, PubSub, Topic

topic_one = Topic("one")
chain_one = Topic.IN.subscribe() | (lambda x: x + 'b') | topic_one.publish()
chain_two = topic_one.subscribe() | (lambda x: x + 'c') | Topic.OUT.publish()

conn = InMemoryPubSubConnection()
pubsub = PubSub(processes=(chain_one, chain_two), connection=conn)

assert pubsub.invoke('a') == ['abc']

Check tests and examples for more examples.

Near-term Roadmap

• Add initial retry support (pending changes in langchain)
• Implement OUT as regular topic
• Implement IN as regular topic
• Add Connection.peek() to monitor past messages from all topics
• Enable resuming PubSub from the "middle" of the computation
• Add test for .peek()
• Add "wait until topic X is done" pattern, aka. Topic.join()
• Move tracking of inflight processes/messages to Connection
  • Use this to build retry mechanism, where any inflight messages are moved back to the respective topics when restarting
  • But this would require being able to replay a message for a single listener only, which maybe requires a larger redesign of PubSub<>Connection contract than what I wanted to do here
• Detect cycles (aka. infinite loops) and throw an error
  • Allow user to catch that error (by subcribing to an error topic?)
• Add example for "human in the loop" pattern, one of the two below
  • Example with one permchain, which runs until it produces either 1. request for input or 2. output. The consumer code then gets the needed info, and restarts the permchain with answer, and same state id
  • Allow interrupting execution by breaking out of the iterator returned by .stream()
    • Build example showing a simple "human in the loop" pattern using this, ie. if a certain message asking for input is published the consumer of the iterator breaks out, does something and then restarts it
• Add Redis-backed Connection implementation