Add some intro docs to readme

README.md

@@ -4,7 +4,47 @@
 
 `pip install permchain`
 
-## Usage
+## Overview
+
+PermChain is an alpha-stage library for building stateful, multi-actor applications with LLMs. It extends the [LangChain Expression Language](https://python.langchain.com/docs/expression_language/) with the ability to coordinate multiple chains (or actors) across multiple steps of computation. It is inspired by [Pregel](https://research.google/pubs/pub37252/) and [Apache Beam](https://beam.apache.org/).
+
+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:
+
+- `Channels.LastValue`: stores the last value sent to the channel, useful for input values, and single-value outputs
+- `Channels.Inbox`: stores an ephemeral sequence of values sent to the channel, useful for sending data from one chain to another
+- `Channels.UniqueInbox`: same as Inbox, but deduplicates values sent to the channel
+- `Channels.Archive`: stores a persistent sequence of values sent to the channel, useful for accumulating data over multiple steps
+- `Channels.UniqueArchive`: same as Archive, but deduplicates values sent to the channel
+- `Channels.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 = Channels.BinaryOperatorAggregate(int, operator.add)`
+- `Channels.ContextManager`: exposes the value of a context manager, managing its lifecycle. Useful for accessing external resources that require setup and/or teardown. eg. `client = Channels.ContextManager(httpx.Client)`
+
+### 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
 
 ```python
 from permchain import Channels, Pregel

permchain/channels/archive.py

@@ -4,11 +4,11 @@ from typing import Any, FrozenSet, Generator, Generic, Optional, Sequence, Type
 
 from typing_extensions import Self
 
-from permchain.channels.base import Channel, EmptyChannelError, Value
+from permchain.channels.base import BaseChannel, EmptyChannelError, Value
 from permchain.channels.inbox import flatten
 
 
-class Archive(Generic[Value], Channel[Sequence[Value], Value | list[Value]]):
+class Archive(Generic[Value], BaseChannel[Sequence[Value], Value | list[Value]]):
     """Stores all unique values received, persists across steps."""
 
     def __init__(self, typ: Type[Value]) -> None:
@@ -51,7 +51,7 @@ class Archive(Generic[Value], Channel[Sequence[Value], Value | list[Value]]):
             raise EmptyChannelError()
 
 
-class UniqueArchive(Generic[Value], Channel[FrozenSet[Value], Value]):
+class UniqueArchive(Generic[Value], BaseChannel[FrozenSet[Value], Value]):
     """Stores all unique values received, persists across steps."""
 
     def __init__(self, typ: Type[Value]) -> None:

permchain/channels/base.py

@@ -30,7 +30,7 @@ class InvalidUpdateError(Exception):
     pass
 
 
-class Channel(Generic[Value, Update], ABC):
+class BaseChannel(Generic[Value, Update], ABC):
     @property
     @abstractmethod
     def ValueType(self) -> Any:
@@ -77,8 +77,8 @@ class Channel(Generic[Value, Update], ABC):
 
 @contextmanager
 def ChannelsManager(
-    channels: Mapping[str, Channel]
-) -> Generator[Mapping[str, Channel], None, None]:
+    channels: Mapping[str, BaseChannel]
+) -> Generator[Mapping[str, BaseChannel], None, None]:
     """Manage channels for the lifetime of a Pregel invocation (multiple steps)."""
     empty = {k: v.empty() for k, v in channels.items()}
     try:
@@ -90,8 +90,8 @@ def ChannelsManager(
 
 @asynccontextmanager
 async def AsyncChannelsManager(
-    channels: Mapping[str, Channel]
-) -> AsyncGenerator[Mapping[str, Channel], None]:
+    channels: Mapping[str, BaseChannel]
+) -> AsyncGenerator[Mapping[str, BaseChannel], None]:
     """Manage channels for the lifetime of a Pregel invocation (multiple steps)."""
     empty = {k: v.aempty() for k, v in channels.items()}
     try:

permchain/channels/binop.py

@@ -4,16 +4,16 @@ from typing import Callable, Generator, Generic, Optional, Sequence, Type
 
 from typing_extensions import Self
 
-from permchain.channels.base import Channel, EmptyChannelError, Value
+from permchain.channels.base import BaseChannel, EmptyChannelError, Value
 
 
-class BinaryOperatorAggregate(Generic[Value], Channel[Value, Value]):
+class BinaryOperatorAggregate(Generic[Value], BaseChannel[Value, Value]):
     """Stores the result of applying a binary operator to the current value and each new value.
 
     ```python
     import operator
 
-    total = BinaryOperatorAggregate(int, operator.add)
+    total = Channels.BinaryOperatorAggregate(int, operator.add)
     ```
     """
 

permchain/channels/context.py

@@ -15,14 +15,14 @@ from typing import ContextManager as ContextManagerType
 from typing_extensions import Self
 
 from permchain.channels.base import (
-    Channel,
+    BaseChannel,
     EmptyChannelError,
     InvalidUpdateError,
     Value,
 )
 
 
-class ContextManager(Generic[Value], Channel[Value, None]):
+class ContextManager(Generic[Value], BaseChannel[Value, None]):
     """Exposes the value of a context manager, for the duration of an invocation.
     Context manager is entered before the first step, and exited after the last step.
     Optionally, provide an equivalent async context manager, which will be used
@@ -31,7 +31,7 @@ class ContextManager(Generic[Value], Channel[Value, None]):
     ```python
     import httpx
 
-    client = ContextManager(httpx.Client, httpx.AsyncClient)
+    client = Channels.ContextManager(httpx.Client, httpx.AsyncClient)
     ```
     """
 

permchain/channels/inbox.py

@@ -14,7 +14,7 @@ from typing import (
 
 from typing_extensions import Self
 
-from permchain.channels.base import Channel, EmptyChannelError, Value
+from permchain.channels.base import BaseChannel, EmptyChannelError, Value
 
 
 def flatten(values: Sequence[Value | list[Value]]) -> Iterator[Value]:
@@ -25,7 +25,7 @@ def flatten(values: Sequence[Value | list[Value]]) -> Iterator[Value]:
             yield value
 
 
-class Inbox(Generic[Value], Channel[Sequence[Value], Value | list[Value]]):
+class Inbox(Generic[Value], BaseChannel[Sequence[Value], Value | list[Value]]):
     """Stores all values received, resets in each step."""
 
     def __init__(self, typ: Type[Value]) -> None:
@@ -70,7 +70,7 @@ class Inbox(Generic[Value], Channel[Sequence[Value], Value | list[Value]]):
             raise EmptyChannelError()
 
 
-class UniqueInbox(Generic[Value], Channel[FrozenSet[Value], Value | list[Value]]):
+class UniqueInbox(Generic[Value], BaseChannel[FrozenSet[Value], Value | list[Value]]):
     """Stores all unique values received, resets in each step."""
 
     def __init__(self, typ: Type[Value]) -> None:

permchain/channels/last_value.py

@@ -5,14 +5,14 @@ from typing import Generator, Generic, Optional, Sequence, Type
 from typing_extensions import Self
 
 from permchain.channels.base import (
-    Channel,
+    BaseChannel,
     EmptyChannelError,
     InvalidUpdateError,
     Value,
 )
 
 
-class LastValue(Generic[Value], Channel[Value, Value]):
+class LastValue(Generic[Value], BaseChannel[Value, Value]):
     """Stores the last value received, can receive at most one value per step."""
 
     def __init__(self, typ: Type[Value]) -> None:

permchain/pregel/__init__.py

@@ -35,7 +35,7 @@ from langchain.schema.runnable.config import (
 
 from permchain.channels.base import (
     AsyncChannelsManager,
-    Channel,
+    BaseChannel,
     ChannelsManager,
     EmptyChannelError,
 )
@@ -49,7 +49,7 @@ from permchain.pregel.write import PregelSink
 
 
 class Pregel(RunnableSerializable[dict[str, Any] | Any, dict[str, Any] | Any]):
-    channels: Mapping[str, Channel]
+    channels: Mapping[str, BaseChannel]
 
     chains: Mapping[str, PregelInvoke | PregelBatch]
 
@@ -403,7 +403,7 @@ def _interrupt_or_proceed(
 
 def _apply_writes_and_prepare_next_tasks(
     processes: Mapping[str, PregelInvoke | PregelBatch],
-    channels: Mapping[str, Channel],
+    channels: Mapping[str, BaseChannel],
     pending_writes: Sequence[tuple[str, Any]],
 ) -> list[tuple[Runnable, Any, str]]:
     pending_writes_by_channel: dict[str, list[Any]] = defaultdict(list)

permchain/pregel/debug.py

@@ -4,7 +4,7 @@ from typing import Any, Iterator, Mapping
 from langchain.schema.runnable import Runnable
 from langchain.utils.input import get_bolded_text, get_colored_text
 
-from permchain.channels.base import Channel, EmptyChannelError
+from permchain.channels.base import BaseChannel, EmptyChannelError
 
 
 def print_step_start(step: int, next_tasks: list[tuple[Runnable, Any, str]]) -> None:
@@ -18,7 +18,7 @@ def print_step_start(step: int, next_tasks: list[tuple[Runnable, Any, str]]) ->
     )
 
 
-def print_checkpoint(step: int, channels: Mapping[str, Channel]) -> None:
+def print_checkpoint(step: int, channels: Mapping[str, BaseChannel]) -> None:
     print(
         f"{get_colored_text('[pregel/checkpoint]', color='blue')} "
         + get_bolded_text(f"Finishing step {step}. Channel values:\n")
@@ -26,7 +26,7 @@ def print_checkpoint(step: int, channels: Mapping[str, Channel]) -> None:
     )
 
 
-def _read_channels(channels: Mapping[str, Channel]) -> Iterator[tuple[str, Any]]:
+def _read_channels(channels: Mapping[str, BaseChannel]) -> Iterator[tuple[str, Any]]:
     for name, channel in channels.items():
         try:
             yield (name, channel.get())

permchain/pregel/io.py

@@ -1,6 +1,6 @@
 from typing import Any, Iterator, Mapping, Sequence
 
-from permchain.channels.base import Channel
+from permchain.channels.base import BaseChannel
 from permchain.pregel.log import logger
 
 
@@ -23,7 +23,7 @@ def map_input(
 def map_output(
     output_channels: str | Sequence[str],
     pending_writes: Sequence[tuple[str, Any]],
-    channels: Mapping[str, Channel],
+    channels: Mapping[str, BaseChannel],
 ) -> Iterator[dict[str, Any] | Any]:
     """Map pending writes (a sequence of tuples (channel, value)) to output chunk."""
     if isinstance(output_channels, str):

permchain/pregel/read.py

@@ -13,7 +13,7 @@ from langchain.schema.runnable import (
 from langchain.schema.runnable.base import Other, RunnableEach, coerce_to_runnable
 from langchain.schema.runnable.utils import ConfigurableFieldSpec
 
-from permchain.channels.base import Channel
+from permchain.channels.base import BaseChannel
 from permchain.pregel.constants import CONFIG_KEY_READ
 
 
@@ -28,7 +28,7 @@ class PregelRead(RunnableLambda):
                 name=CONFIG_KEY_READ,
                 description=None,
                 default=None,
-                annotation=Callable[[Channel], Any],
+                annotation=Callable[[BaseChannel], Any],
             ),
         ]
 

permchain/pregel/validate.py

@@ -1,12 +1,12 @@
 from typing import Mapping, Sequence
 
-from permchain.channels.base import Channel
+from permchain.channels.base import BaseChannel
 from permchain.pregel.read import PregelBatch, PregelInvoke
 
 
 def validate_chains_channels(
     chains: Mapping[str, PregelInvoke | PregelBatch],
-    channels: Mapping[str, Channel],
+    channels: Mapping[str, BaseChannel],
     input: str | Sequence[str],
     output: str | Sequence[str],
 ) -> None: