From 14d84825da0becf2af5f1231ae301463227e0432 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Mon, 23 Oct 2023 15:06:07 +0100 Subject: [PATCH] Add some intro docs to readme --- README.md | 42 +++++++++++++++++++++++++++++++- permchain/channels/archive.py | 6 ++--- permchain/channels/base.py | 10 ++++---- permchain/channels/binop.py | 6 ++--- permchain/channels/context.py | 6 ++--- permchain/channels/inbox.py | 6 ++--- permchain/channels/last_value.py | 4 +-- permchain/pregel/__init__.py | 6 ++--- permchain/pregel/debug.py | 6 ++--- permchain/pregel/io.py | 4 +-- permchain/pregel/read.py | 4 +-- permchain/pregel/validate.py | 4 +-- 12 files changed, 72 insertions(+), 32 deletions(-) diff --git a/README.md b/README.md index 3206e51a4..e8c919d12 100644 --- a/README.md +++ b/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 diff --git a/permchain/channels/archive.py b/permchain/channels/archive.py index 05015fbc4..a491e9971 100644 --- a/permchain/channels/archive.py +++ b/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: diff --git a/permchain/channels/base.py b/permchain/channels/base.py index 789d7e2af..511df1d43 100644 --- a/permchain/channels/base.py +++ b/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: diff --git a/permchain/channels/binop.py b/permchain/channels/binop.py index 9861c85e5..8f125673b 100644 --- a/permchain/channels/binop.py +++ b/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) ``` """ diff --git a/permchain/channels/context.py b/permchain/channels/context.py index 87664aacb..a95f2eff7 100644 --- a/permchain/channels/context.py +++ b/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) ``` """ diff --git a/permchain/channels/inbox.py b/permchain/channels/inbox.py index f49ae8887..60972a79b 100644 --- a/permchain/channels/inbox.py +++ b/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: diff --git a/permchain/channels/last_value.py b/permchain/channels/last_value.py index d60b064d6..953d16dad 100644 --- a/permchain/channels/last_value.py +++ b/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: diff --git a/permchain/pregel/__init__.py b/permchain/pregel/__init__.py index b8e49a822..11ff4f16e 100644 --- a/permchain/pregel/__init__.py +++ b/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) diff --git a/permchain/pregel/debug.py b/permchain/pregel/debug.py index 636512bcc..613749858 100644 --- a/permchain/pregel/debug.py +++ b/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()) diff --git a/permchain/pregel/io.py b/permchain/pregel/io.py index ddb0a4688..460554877 100644 --- a/permchain/pregel/io.py +++ b/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): diff --git a/permchain/pregel/read.py b/permchain/pregel/read.py index 87399c78c..709b4e6b8 100644 --- a/permchain/pregel/read.py +++ b/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], ), ] diff --git a/permchain/pregel/validate.py b/permchain/pregel/validate.py index bdc201c49..bd269845c 100644 --- a/permchain/pregel/validate.py +++ b/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: