diff --git a/examples/agents/75_wait_for_message.py b/examples/agents/75_wait_for_message.py index 42c8b17f..e0a2276d 100644 --- a/examples/agents/75_wait_for_message.py +++ b/examples/agents/75_wait_for_message.py @@ -3,11 +3,17 @@ Demonstrates: - wait_for_message_tool: dequeues messages from the WMQ (Conductor PULL_WORKFLOW_MESSAGES task) - Mixing a server-side message tool with a local Python action tool - - Looping agent that keeps processing messages indefinitely + - Open-ended looping agent: it never decides to stop on its own - Pushing messages from outside the workflow with runtime.send_message() + - handle.stop() ending the loop deterministically -The agent loops forever: each iteration waits for a message, reads the -"task" field, executes it, and goes back to listening. +Each iteration waits for a message, reads the "task" field, executes it, and +goes back to listening. The agent's instructions tell it to never stop, so the +loop only ends when the caller calls handle.stop(): that sets the +``_stop_requested`` workflow variable checked by the DoWhile condition and +pushes a ``{"_signal": "stop"}`` message to unblock the pending +PULL_WORKFLOW_MESSAGES. The workflow ends COMPLETED, not TERMINATED — see +84_deterministic_stop.py. Requirements: - Conductor server with WMQ support (conductor.workflow-message-queue.enabled=true) @@ -67,6 +73,7 @@ def main() -> None: # Let the agent process all messages (~5-6s per message) time.sleep(30) + # The agent will never stop on its own — end the loop from here. handle.stop() handle.join(timeout=30) print("\nDone.") diff --git a/examples/agents/76_wait_for_message_streaming.py b/examples/agents/76_wait_for_message_streaming.py index 7f925a09..e9b14add 100644 --- a/examples/agents/76_wait_for_message_streaming.py +++ b/examples/agents/76_wait_for_message_streaming.py @@ -4,10 +4,18 @@ - wait_for_message_tool with streaming: push messages in and see the agent react - Using handle.stream() to observe WAITING → processing → WAITING cycles - runtime.send_message() to push payloads into the Workflow Message Queue + - handle.stop() ending the loop deterministically -The agent starts, immediately waits for a message, processes whatever it -receives (by calling wait_for_message again), then waits again. The caller -drives the conversation by sending messages and reading streamed events. +The agent starts, immediately waits for a message, answers it with respond(), +then loops back to wait_for_message. The caller drives the conversation from a +background thread — sending a task every 8 seconds — while the main thread reads +streamed events. + +The agent's instructions tell it to never stop, so the loop only ends when the +sender calls handle.stop() — after giving the last task time to be answered. +That sets the ``_stop_requested`` workflow variable +checked by the DoWhile condition and pushes a ``{"_signal": "stop"}`` message to +unblock the pending PULL_WORKFLOW_MESSAGES. stream() then yields DONE. Requirements: - Conductor server running at http://localhost:8080 @@ -66,15 +74,16 @@ def main() -> None: print(f"Agent started: {handle.execution_id}\n") # Push messages from a background thread while we stream events on the main thread. - # Wait long enough between sends for the agent to finish processing each message. - # No sleep after the last send — handle.stream() on the main thread is already the - # barrier: it blocks until DONE, which only fires once the workflow reaches a - # terminal state (after stop() sets the flag and the current iteration completes). + # Wait long enough between sends for the agent to finish processing each message — + # including after the last one. Calling stop() immediately after the final send + # would set _stop_requested while the agent is still mid-turn on that task, and the + # DoWhile would exit before it ever answers. def sender(): for task in TASKS: time.sleep(8) print(f"\n [caller] sending -> {task!r}") runtime.send_message(handle.execution_id, {"task": task}) + time.sleep(8) handle.stop() threading.Thread(target=sender, daemon=True).start() diff --git a/examples/agents/77_kafka_consumer_agent.py b/examples/agents/77_kafka_consumer_agent.py index 7ed72ddd..72a970e8 100644 --- a/examples/agents/77_kafka_consumer_agent.py +++ b/examples/agents/77_kafka_consumer_agent.py @@ -58,36 +58,48 @@ def echo_message(value: str, topic: str, offset: int) -> str: ) -with AgentRuntime() as runtime: - handle = runtime.start(agent, "Start consuming messages from Kafka.") - print(f"Agent started: {handle.execution_id}") - - consumer = Consumer( - { - "bootstrap.servers": KAFKA_BOOTSTRAP, - "group.id": KAFKA_GROUP, - "auto.offset.reset": "latest", - } - ) - consumer.subscribe([KAFKA_TOPIC]) - try: - while True: - msg = consumer.poll(timeout=1.0) - if msg is None: - continue - if msg.error(): - if msg.error().code() == KafkaError._PARTITION_EOF: +def main() -> None: + with AgentRuntime() as runtime: + handle = runtime.start(agent, "Start consuming messages from Kafka.") + print(f"Agent started: {handle.execution_id}") + + consumer = Consumer( + { + "bootstrap.servers": KAFKA_BOOTSTRAP, + "group.id": KAFKA_GROUP, + "auto.offset.reset": "latest", + } + ) + consumer.subscribe([KAFKA_TOPIC]) + print(f"Consuming '{KAFKA_TOPIC}' from {KAFKA_BOOTSTRAP} — Ctrl+C to stop.") + try: + while True: + msg = consumer.poll(timeout=1.0) + if msg is None: continue - raise RuntimeError(f"Kafka error: {msg.error()}") - runtime.send_message( - handle.execution_id, - { - "topic": msg.topic(), - "partition": msg.partition(), - "offset": msg.offset(), - "key": msg.key().decode("utf-8") if msg.key() else None, - "value": msg.value().decode("utf-8") if msg.value() else "", - }, - ) - finally: - consumer.close() + if msg.error(): + if msg.error().code() == KafkaError._PARTITION_EOF: + continue + raise RuntimeError(f"Kafka error: {msg.error()}") + runtime.send_message( + handle.execution_id, + { + "topic": msg.topic(), + "partition": msg.partition(), + "offset": msg.offset(), + "key": msg.key().decode("utf-8") if msg.key() else None, + "value": msg.value().decode("utf-8") if msg.value() else "", + }, + ) + except KeyboardInterrupt: + print("\nStopping agent...") + handle.stop() + finally: + consumer.close() + + +# Guard the runtime block: spawned tool workers re-import this module, and +# without the guard they would re-run the orchestration (multiprocessing's +# "Safe importing of main module" error). +if __name__ == "__main__": + main()