permchain 0.0.8

Description:

permchain 0.0.8

permchain
Get started
pip install permchain
Overview
PermChain is an alpha-stage library for building stateful, multi-actor applications with LLMs. It extends the LangChain Expression Language with the ability to coordinate multiple chains (or actors) across multiple steps of computation. It is inspired by Pregel and Apache Beam.
Some of the use cases are:

Recursive/iterative LLM chains
LLM chains with persistent state/memory
LLM agents
Multi-agent simulations
...and more!

How it works
Channels
Channels are used to communicate between chains. Each channel has a value type, an update type, and an update function – which takes a sequence of updates and modifies the stored value. Channels can be used to send data from one chain to another, or to send data from a chain to itself in a future step. PermChain provides a number of built-in channels:
Basic channels: LastValue and Topic

LastValue: The default channel, stores the last value sent to the channel, useful for input and output values, or for sending data from one step to the next
Topic: A configurable PubSub Topic, useful for sending multiple values between chains, or for accumulating output. Can be configured to deduplicate values, and/or to accummulate values over the course of multiple steps.

Advanced channels: Context and BinaryOperatorAggregate

Context: exposes the value of a context manager, managing its lifecycle. Useful for accessing external resources that require setup and/or teardown. eg. client = Context(httpx.Client)
BinaryOperatorAggregate: stores a persistent value, updated by applying a binary operator to the current value and each update sent to the channel, useful for computing aggregates over multiple steps. eg. total = BinaryOperatorAggregate(int, operator.add)

Chains
Chains are LCEL Runnables which subscribe to one or more channels, and write to one or more channels. Any valid LCEL expression can be used as a chain. Chains can be combined into a Pregel application, which coordinates the execution of the chains across multiple steps.
Pregel
Pregel combines multiple chains (or actors) into a single application. It coordinates the execution of the chains across multiple steps, following the Pregel/Bulk Synchronous Parallel model. Each step consists of three phases:

Plan: Determine which chains to execute in this step, ie. the chains that subscribe to channels updated in the previous step (or, in the first step, chains that subscribe to input channels)
Execution: Execute those chains in parallel, until all complete, or one fails, or a timeout is reached. Any channel updates are invisible to other chains until the next step.
Update: Update the channels with the values written by the chains in this step.

Repeat until no chains are planned for execution, or a maximum number of steps is reached.
Example
from permchain import Channel, Pregel

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

app = Pregel(
chains={"grow_value": grow_value},
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
Add an optional Diff value for Channels that implements __add__, returned by update(), yielded by Pregel for output channels. Add replacing_keys set to AddableDict. use an addabledict for yielding values. channels that dont implement it get marked with replacing_keys
Implement checkpointing

Save checkpoints at end of each step/run
Load checkpoint at start of invocation
API to specify storage backend and save key
Tests


Add more examples

multi agent simulation
human in the loop
combine documents
agent executor (add current v total iterations info to read/write steps to enable doing a final update at the end)
run over dataset


Fault tolerance

Expose a unique id to each step, hash of (app, chain, checkpoint) (include input updates for first step)
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}