From 24451a5ed7e3fbad787374e435fea2a67b759e9c Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Tue, 12 Sep 2023 19:23:59 +0100 Subject: [PATCH] Use explicit state_id --- README.md | 5 ++- permchain/pubsub.py | 4 +-- tests/test_invoke.py | 86 ++++++++++++++++++++++---------------------- 3 files changed, 50 insertions(+), 45 deletions(-) diff --git a/README.md b/README.md index c18df4c87..13c088005 100644 --- a/README.md +++ b/README.md @@ -33,6 +33,9 @@ Check `tests` and `examples` for more examples. - [ ] Use this to build retry mechanism, where any inflight messages are moved back to the respective topics when restarting - [ ] Detect cycles (aka. infinite loops) and throw an error - [ ] Allow user to catch that error (by subcribing to an error topic?) -- [ ] Add "human in the loop" pattern +- [ ] Add "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 "wait until topic X is done" pattern, aka. `Topic.reduce()` - [ ] Add Redis-backed Connection implementation diff --git a/permchain/pubsub.py b/permchain/pubsub.py index 6c1aaf402..933e14599 100644 --- a/permchain/pubsub.py +++ b/permchain/pubsub.py @@ -50,8 +50,8 @@ class PubSub(Serializable, Runnable[Any, Any], ABC): input_value += chunk with get_executor_for_config(config) as executor: - # Namespace topics for each run - topic_prefix = str(run_manager.parent_run_id or run_manager.run_id) + # Namespace topics for each run, default to run_id, ie. isolate runs + topic_prefix = str(config.get("state_id") or run_manager.run_id) # Track inflight futures inflight: Set[Future] = set() # Track exceptions diff --git a/tests/test_invoke.py b/tests/test_invoke.py index 3bad75f7a..83713bd2b 100644 --- a/tests/test_invoke.py +++ b/tests/test_invoke.py @@ -1,7 +1,8 @@ +from uuid import uuid4 + import pytest from pytest_mock import MockerFixture -from langchain.callbacks.manager import trace_as_chain_group from permchain.connection_inmemory import InMemoryPubSubConnection from permchain.pubsub import PubSub from permchain.topic import RunnableSubscriber, Topic @@ -67,51 +68,52 @@ def test_invoke_two_processes_in_out_interrupt(mocker: MockerFixture): assert conn.listeners == {} # Then invoke both pubsubs, as a group # The second picks up where the first left off - with trace_as_chain_group("PubSubGroup") as cm: - # invoke() step 1 - assert pubsub_one.invoke(2, {"callbacks": cm}) == [] + state_id = uuid4() - # listeners are still cleared, even though state is preserved - assert conn.listeners == {} - # The log contains all messages published to all topics, in order - assert [{**m, "started_at": None} for m in conn.peek(cm.parent_run_id)] == [ - {"message": 2, "topic_name": "__in__", "started_at": None}, - {"message": 3, "topic_name": "one", "started_at": None}, - ] - # IN, OUT, one - assert len(conn.topics) == 3 - topic_one_full_name = conn.full_topic_name(cm.parent_run_id, topic_one.name) - # the actual message publishd by chain_one, and a sentinel "end" value - assert conn.topics[topic_one_full_name].qsize() == 2 + # invoke() step 1 + assert pubsub_one.invoke(2, {"state_id": state_id}) == [] - # invoke() step 2 - # this picks up where the first left off, and produces same result as - # `test_invoke_two_processes_in_out` - assert pubsub_two.invoke(None, {"callbacks": cm}) == [4] + # listeners are still cleared, even though state is preserved + assert conn.listeners == {} + # The log contains all messages published to all topics, in order + assert [{**m, "started_at": None} for m in conn.peek(state_id)] == [ + {"message": 2, "topic_name": "__in__", "started_at": None}, + {"message": 3, "topic_name": "one", "started_at": None}, + ] + # IN, OUT, one + assert len(conn.topics) == 3 + topic_one_full_name = conn.full_topic_name(state_id, topic_one.name) + # the actual message publishd by chain_one, and a sentinel "end" value + assert conn.topics[topic_one_full_name].qsize() == 2 - # listeners are still cleared, even though state is preserved - assert conn.listeners == {} - # The log contains all messages published to all topics, in order - assert [{**m, "started_at": None} for m in conn.peek(cm.parent_run_id)] == [ - {"message": 2, "topic_name": "__in__", "started_at": None}, - {"message": 3, "topic_name": "one", "started_at": None}, - {"message": None, "topic_name": "__in__", "started_at": None}, - {"message": 4, "topic_name": "__out__", "started_at": None}, - ] - # IN, OUT, one - assert len(conn.topics) == 3 + # invoke() step 2 + # this picks up where the first left off, and produces same result as + # `test_invoke_two_processes_in_out` + assert pubsub_two.invoke(None, {"state_id": state_id}) == [4] - for topic_name, queue in conn.topics.items(): - if topic_name.endswith("IN"): - # Contains two sentinel "end" values, and None - # passed in as input to chain_two, which doesn't subscribe to it - assert queue.qsize() == 3 - if topic_name.endswith("OUT"): - # Empty because this was consumed by invoke() - assert queue.qsize() == 0 - if topic_name.endswith("one"): - # Contains sentinel "end" value - assert queue.qsize() == 1 + # listeners are still cleared, even though state is preserved + assert conn.listeners == {} + # The log contains all messages published to all topics, in order + assert [{**m, "started_at": None} for m in conn.peek(state_id)] == [ + {"message": 2, "topic_name": "__in__", "started_at": None}, + {"message": 3, "topic_name": "one", "started_at": None}, + {"message": None, "topic_name": "__in__", "started_at": None}, + {"message": 4, "topic_name": "__out__", "started_at": None}, + ] + # IN, OUT, one + assert len(conn.topics) == 3 + + for topic_name, queue in conn.topics.items(): + if topic_name.endswith("IN"): + # Contains two sentinel "end" values, and None + # passed in as input to chain_two, which doesn't subscribe to it + assert queue.qsize() == 3 + if topic_name.endswith("OUT"): + # Empty because this was consumed by invoke() + assert queue.qsize() == 0 + if topic_name.endswith("one"): + # Contains sentinel "end" value + assert queue.qsize() == 1 def test_invoke_many_processes_in_out(mocker: MockerFixture):