diff --git a/examples/agents/78_approval_workflow.py b/examples/agents/78_approval_workflow.py index 91d9d7fb..71b8bc1a 100644 --- a/examples/agents/78_approval_workflow.py +++ b/examples/agents/78_approval_workflow.py @@ -1,7 +1,8 @@ """Approval Workflow — agent dynamically decides which tasks need human sign-off. Demonstrates: - - wait_for_message_tool as a dynamic approval gate driven by LLM reasoning + - wait_for_message_tool as the task intake; flag_for_approval (a @tool) as a + dynamic approval gate driven by LLM reasoning - The agent itself decides mid-loop whether a task is risky, rather than the workflow being designed with an explicit approval step upfront - flag_for_approval blocks until the operator decides, returning "approve" @@ -9,12 +10,15 @@ which prevents the agent from pulling the next task while approval is pending - Filesystem-based IPC between the main process and worker processes: tool workers run as separate OS processes (different PIDs, same filesystem), - so @tool functions use sentinel files to communicate with the main thread - - Clean shutdown: the agent responds with no tool calls on the stop signal, - which lets the DoWhile loop exit naturally (workflow ends COMPLETED) - -How this differs from examples 09a–09d (HITL): - In 09a–09d the approval pause is a WaitTask node baked into the workflow + so @tool functions use sentinel files to talk to the main process. The + shared directory crosses process boundaries via APPROVAL_WORKFLOW_IPC_DIR — + a per-import mkdtemp() would give every worker its own dir. + - Deterministic stop: handle.stop() ends the loop once every task has been + accounted for, without any stop-handling instructions in the prompt + (workflow ends COMPLETED) + +How this differs from examples 09–09d (HITL): + In 09–09d the approval pause is a WaitTask node baked into the workflow definition at compile time — the workflow always pauses at that point regardless of the input. Here, the LLM inspects each incoming task and decides dynamically whether it is safe to execute immediately or requires @@ -31,7 +35,7 @@ blocks on flag_for_approval until the operator responds. Requirements: - - Conductor server running at http://localhost:8080 + - Conductor server with WMQ support (conductor.workflow-message-queue.enabled=true) - CONDUCTOR_SERVER_URL=http://localhost:8080/api as environment variable - CONDUCTOR_AGENT_LLM_MODEL=openai/gpt-4o-mini as environment variable """ @@ -49,12 +53,19 @@ from settings import settings # Shared directory for IPC between main process and worker processes. -# Workers run as separate OS processes (different PIDs, same filesystem). -_ipc_dir = Path(tempfile.mkdtemp(prefix="approval_workflow_")) +# Workers run as separate OS processes (different PIDs, same filesystem) that +# re-import this module, so the directory is passed down via an env var — a +# per-import mkdtemp() would give every worker its own dir and break the IPC. +_IPC_DIR_ENV = "APPROVAL_WORKFLOW_IPC_DIR" +if _IPC_DIR_ENV in os.environ: + _ipc_dir = Path(os.environ[_IPC_DIR_ENV]) +else: + _ipc_dir = Path(tempfile.mkdtemp(prefix="approval_workflow_")) + os.environ[_IPC_DIR_ENV] = str(_ipc_dir) _APPROVAL_DIR = _ipc_dir / "approvals" _DONE_DIR = _ipc_dir / "done" -_APPROVAL_DIR.mkdir() -_DONE_DIR.mkdir() +_APPROVAL_DIR.mkdir(exist_ok=True) +_DONE_DIR.mkdir(exist_ok=True) @tool @@ -94,7 +105,7 @@ def log_rejection(task: str) -> str: receive_message = wait_for_message_tool( name="wait_for_message", - description="Dequeue the next task or stop signal ({stop: true}).", + description="Dequeue the next task to process.", ) agent = Agent( @@ -127,35 +138,43 @@ def log_rejection(task: str) -> str: "Grant admin access to user@example.com", ] -try: - with AgentRuntime() as runtime: - handle = runtime.start(agent, "Start processing the task queue.") - execution_id = handle.execution_id - time.sleep(4) - print(f"Agent started: {execution_id}\n") - - print("Dispatching all tasks...\n") - for task in TASKS: - print(f" → {task!r}") - runtime.send_message(execution_id, {"task": task}) - - # Poll for approval requests; write decision files to unblock the tool. - # Poll for completions to know when to send the stop signal. - while len(list(_DONE_DIR.iterdir())) < len(TASKS): - for req in sorted(_APPROVAL_DIR.glob("*.json")): - data = json.loads(req.read_text()) - req.unlink() - print(f"\n ⚠ APPROVAL REQUIRED") - print(f" Task: {data['task']}") - print(f" Reason: {data['reason']}\n") - answer = input(" Approve? [Y/N]: ").strip().upper() - decision = "approve" if answer == "Y" else "reject" - req.with_suffix(".decision").write_text(decision) - time.sleep(0.1) - - # Deterministic stop — no stop-handling instructions needed. - handle.stop() - handle.join(timeout=30) - print("\nDone.") -finally: - shutil.rmtree(_ipc_dir, ignore_errors=True) +def main() -> None: + try: + with AgentRuntime() as runtime: + handle = runtime.start(agent, "Start processing the task queue.") + execution_id = handle.execution_id + time.sleep(4) + print(f"Agent started: {execution_id}\n") + + print("Dispatching all tasks...\n") + for task in TASKS: + print(f" → {task!r}") + runtime.send_message(execution_id, {"task": task}) + + # Poll for approval requests; write decision files to unblock the tool. + # Poll for completions to know when to send the stop signal. + while len(list(_DONE_DIR.iterdir())) < len(TASKS): + for req in sorted(_APPROVAL_DIR.glob("*.json")): + data = json.loads(req.read_text()) + req.unlink() + print("\n ⚠ APPROVAL REQUIRED") + print(f" Task: {data['task']}") + print(f" Reason: {data['reason']}\n") + answer = input(" Approve? [Y/N]: ").strip().upper() + decision = "approve" if answer == "Y" else "reject" + req.with_suffix(".decision").write_text(decision) + time.sleep(0.1) + + # Deterministic stop — no stop-handling instructions needed. + handle.stop() + handle.join(timeout=30) + print("\nDone.") + finally: + shutil.rmtree(_ipc_dir, ignore_errors=True) + + +# 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() diff --git a/examples/agents/79_agent_message_bus.py b/examples/agents/79_agent_message_bus.py index f2dde389..55051c13 100644 --- a/examples/agents/79_agent_message_bus.py +++ b/examples/agents/79_agent_message_bus.py @@ -3,10 +3,16 @@ Demonstrates: - Agent-to-agent messaging: one running agent sending messages directly into another running agent's WMQ via runtime.send_message() - - A tool that closes over an execution_id to forward results downstream + - Module-level tools that pick up runtime values from the environment: + forward_to_writer reads the Writer's execution id from + MESSAGE_BUS_WRITER_EXECUTION_ID, since a tool that closed over it could not + be pickled to its spawned worker process - Parallel agent pipelines: researcher → writer running concurrently - - Filesystem-based IPC: forward_to_writer writes sentinel files so the main - thread knows when all topics have been forwarded + - Filesystem-based IPC between the main process and worker processes: + forward_to_writer and publish each write sentinel files, so the main process + can tell forwarding from publishing. The barrier waits on publish — the + Researcher forwards the last topic while the Writer is still mid-turn on it, + so stopping at "all forwarded" would cut the final paragraph. - Deterministic stop: handle.stop() exits each agent's loop gracefully How this differs from 06_sequential_pipeline: @@ -26,7 +32,7 @@ Researcher autonomously drives the Writer. Requirements: - - Conductor server running at http://localhost:8080 + - Conductor server with WMQ support (conductor.workflow-message-queue.enabled=true) - CONDUCTOR_SERVER_URL=http://localhost:8080/api as environment variable - CONDUCTOR_AGENT_LLM_MODEL=openai/gpt-4o-mini as environment variable """ @@ -48,8 +54,10 @@ else: _ipc_dir = Path(tempfile.mkdtemp(prefix="message_bus_")) os.environ[_IPC_DIR_ENV] = str(_ipc_dir) -_FORWARDED_DIR = _ipc_dir / "forwarded" # one file per forwarded topic +_FORWARDED_DIR = _ipc_dir / "forwarded" # one file per topic forwarded by the Researcher _FORWARDED_DIR.mkdir(exist_ok=True) +_PUBLISHED_DIR = _ipc_dir / "published" # one file per paragraph published by the Writer +_PUBLISHED_DIR.mkdir(exist_ok=True) TOPICS = [ "the impact of edge computing on cloud infrastructure", @@ -101,6 +109,7 @@ def publish(topic: str, paragraph: str) -> str: """Publish the finished paragraph.""" print(f"\n [writer] ── {topic} ──") print(f" {paragraph}\n") + (_PUBLISHED_DIR / f"{time.time_ns()}.done").touch() return "published" @@ -152,8 +161,18 @@ def main() -> None: print(f" → {topic!r}") runtime.send_message(researcher_id, {"topic": topic}) - # Wait until all topics have been forwarded to the Writer - while len(list(_FORWARDED_DIR.iterdir())) < len(TOPICS): + # Wait until the Writer has published every paragraph. Gating on + # _FORWARDED_DIR is not enough: the Researcher forwards the last topic + # while the Writer is still mid-turn on it, so stopping there would cut + # the final paragraph and can leave the Researcher's stop() racing an + # in-flight iteration. + deadline = time.monotonic() + 180 + while len(list(_PUBLISHED_DIR.iterdir())) < len(TOPICS): + if time.monotonic() > deadline: + raise TimeoutError( + f"Writer published {len(list(_PUBLISHED_DIR.iterdir()))} of " + f"{len(TOPICS)} paragraphs before the deadline." + ) time.sleep(0.1) # Deterministic stop — no stop-handling instructions needed. diff --git a/examples/agents/80_live_dashboard.py b/examples/agents/80_live_dashboard.py index cdabaefb..e6e525a1 100644 --- a/examples/agents/80_live_dashboard.py +++ b/examples/agents/80_live_dashboard.py @@ -37,7 +37,7 @@ Requirements: - Conductor server running at http://localhost:8080 - CONDUCTOR_SERVER_URL=http://localhost:8080/api as environment variable - - CONDUCTOR_AGENT_LLM_MODEL=anthropic/claude-sonnet-4-20250514 as environment variable + - CONDUCTOR_AGENT_LLM_MODEL=anthropic/claude-sonnet-5 as environment variable """ import json diff --git a/examples/agents/81_chat_repl.py b/examples/agents/81_chat_repl.py index 8c54170a..244d9f15 100644 --- a/examples/agents/81_chat_repl.py +++ b/examples/agents/81_chat_repl.py @@ -15,8 +15,10 @@ communicated via the shared filesystem rather than an in-process queue. Resume support: - The REPL saves the execution_id to a session file on start. On subsequent - runs, pass ``--resume`` to reconnect to the same workflow. ``resume()`` + The REPL saves the execution_id to a session file on start. Leave with + ``/disconnect`` to exit the console without stopping the agent (``quit`` / + ``exit`` stop it), then pass ``--resume`` on a later run to reconnect to the + same workflow. ``resume()`` fetches the workflow from the server, extracts the worker domain from ``taskToDomain``, and re-registers tools under that domain — so stateful agents resume correctly. Conversation history is not restored in the @@ -30,7 +32,7 @@ activate predefined text-processing tasks at runtime. The agent is notified via a WMQ message and can start using the new capability immediately. - Built-in tasks (activate with /tool ): + Built-in tasks (activate with /tool , list the active ones with /tools): word_count — count words in input char_count — count characters in input reverse — reverse the input string @@ -41,9 +43,9 @@ bullet_split — split input into one bullet point per sentence Requirements: - - Conductor server running at http://localhost:8080 + - Conductor server with WMQ support (conductor.workflow-message-queue.enabled=true) - CONDUCTOR_SERVER_URL=http://localhost:8080/api as environment variable - - CONDUCTOR_AGENT_LLM_MODEL=anthropic/claude-sonnet-4-20250514 as environment variable + - CONDUCTOR_AGENT_LLM_MODEL=anthropic/claude-sonnet-5 as environment variable """ import argparse diff --git a/examples/agents/82_coding_agent.py b/examples/agents/82_coding_agent.py index 9ce43de3..0dd4fe53 100644 --- a/examples/agents/82_coding_agent.py +++ b/examples/agents/82_coding_agent.py @@ -15,9 +15,9 @@ python 82_coding_agent.py --resume # resume last session Requirements: - - Conductor server running at http://localhost:8080 + - Conductor server with WMQ support (conductor.workflow-message-queue.enabled=true) - CONDUCTOR_SERVER_URL=http://localhost:8080/api - - CONDUCTOR_AGENT_LLM_MODEL=anthropic/claude-sonnet-4-20250514 + - CONDUCTOR_AGENT_LLM_MODEL=anthropic/claude-sonnet-5 """ import argparse @@ -461,9 +461,12 @@ def build_agent(working_dir: str, shell_timeout: int = _DEFAULT_SHELL_TIMEOUT) - - run_shell(command) run a shell command (cwd: {working_dir}, timeout: {shell_timeout}s) - find_files(pattern, path=".") find files by glob, e.g. "**/*.py" - search_in_files(regex, path=".", file_glob) grep files by regex -- reply_to_user(message) send your response to the user +- reply_to_user(message) REQUIRED — how the user sees your answer Rules: +- You MUST call reply_to_user before calling wait_for_message again. It is the only + channel the user can see — plain text replies are discarded. +- Never call wait_for_message twice in a row. Every task ends with reply_to_user. - Work autonomously. Do not ask for permission before reading files, running commands, or writing. - Make as many tool calls as needed to fully complete the task before replying. - Keep replies concise: what was done, what changed, key output. No lengthy explanations. @@ -474,8 +477,8 @@ def build_agent(working_dir: str, shell_timeout: int = _DEFAULT_SHELL_TIMEOUT) - 1. Call wait_for_message to receive the next task. 2. Think through the task. Explore, read, search, modify, and run as needed. 3. Complete the task fully. -4. Call reply_to_user with a concise summary. -5. Return to step 1 immediately. +4. Call reply_to_user with a concise summary. Never skip this step. +5. Only then return to step 1. """, ) diff --git a/examples/agents/82_fan_out_fan_in.py b/examples/agents/82_fan_out_fan_in.py index 3fe65431..dc32726c 100644 --- a/examples/agents/82_fan_out_fan_in.py +++ b/examples/agents/82_fan_out_fan_in.py @@ -21,7 +21,9 @@ - Filesystem IPC: * Workers write sentinels after submit_answer so main counts completions * Collector writes reports to files; main thread reads and prints them - - No time.sleep() to assume message delivery — synchronisation via files + - Result delivery is never assumed from elapsed time — every wait is on a + sentinel file. The one bare sleep is a startup grace period, letting the + agents reach their first wait call before the first message is sent. Scenario: A research Orchestrator fans out each question to three Worker agents @@ -29,8 +31,9 @@ aggregates the three answers into a side-by-side comparison report. Requirements: - - Conductor server running (CONDUCTOR_SERVER_URL / CONDUCTOR_SERVER_URL) - - CONDUCTOR_AGENT_LLM_MODEL set to a working model + - Conductor server with WMQ support (conductor.workflow-message-queue.enabled=true) + - CONDUCTOR_SERVER_URL=http://localhost:8080/api as environment variable + - CONDUCTOR_AGENT_LLM_MODEL=openai/gpt-4o-mini as environment variable """ import json diff --git a/examples/agents/84_deterministic_stop.py b/examples/agents/84_deterministic_stop.py index be3c78a9..a5bcb8c3 100644 --- a/examples/agents/84_deterministic_stop.py +++ b/examples/agents/84_deterministic_stop.py @@ -8,8 +8,8 @@ How it works: The server compiles every agent's DoWhile loop with a ``_stop_requested`` - workflow variable in its condition. When ``handle.stop()`` is called, the - SDK sets this variable to ``true`` via Conductor's ``updateVariables`` API. + workflow variable in its condition. ``handle.stop()`` POSTs to + ``/agent/{execution_id}/stop`` and the server sets that variable to ``true``. The loop condition evaluates to ``false`` on the next check, and the loop exits. The LLM cannot override this — it's checked by Conductor, not the LLM. @@ -28,7 +28,8 @@ The LLM could ignore this. handle.stop() makes this unnecessary. Requirements: - - Conductor server (with _stop_requested support in compiler) + - Conductor server with WMQ support (conductor.workflow-message-queue.enabled=true) + and _stop_requested support in the compiler - CONDUCTOR_SERVER_URL=http://localhost:8080/api as environment variable - CONDUCTOR_AGENT_LLM_MODEL=openai/gpt-4o-mini as environment variable """