jiuwenswarm

How to keep an AI workflow predictable using WorkSwarm's SwarmFlow Have you ever watched one early mistake spread through a whole team of AI agents? SwarmFlow is built to stop that. It is WorkSwarm's scripting layer for Cluster mode (‘team’ mode). You fix the order of steps in a Python file, hold the run at a human checkpoint, and cap token spend with two separate limits. Every run then follows the same steps in the same order, and leaves a record you can audit. Here is the problem it solves. When a team of LLM agents works through a multi-step job, a bad output early in the chain gets passed to every step after it. Say a code-generation agent writes the wrong function. A documentation agent then documents that wrong function. You pay for both calls before anyone looks at the result. There is a second problem. The Leader agent (the one that coordinates the worker agents) decides the order of work at run time. Two runs of the same request can take different paths. WorkSwarm (formerly JiuwenSwarm) tackles this with SwarmFlow. A Python file fixes the order of stages. A human() call holds the run until a person answers. Two token ceilings stop a run and send the Leader a diagnostic message it can act on. In this tutorial, you will: - Build a four-stage pipeline: spec, codegen, human review, and docs. - Send an edit request at the review checkpoint. Then watch SwarmFlow re-run only the changed stage and replay the unchanged one from its journal (a saved record of each finished step). - Trip both token ceilings on purpose. Every output shown here comes from a proof test that ran all four scenarios three times against OpenAI gpt-5.5. Prerequisites and setup You will need: - Linux or macOS with Python 3.11 to 3.13 (the proof ran Python 3.12 on Ubuntu) - An OpenAI API key with access to gpt-5.5 - Ports 19000, 19001 and 18092 free (WorkSwarm's web channel, gateway and agent server) WorkSwarm is published on PyPI as workswarm, because the jiuwenswarm PyPI name stopped at 0.2.3. The workswarm package installs the jiuwenswarm-* commands and pulls in openJiuwen Agent Core (openjiuwen), which contains the SwarmFlow engine. Pin both versions in requirements.txt: workswarm==0.2.7 openjiuwen==0.1.19 Clone the repo and run the proof: git clone https://github.com/manishh/openjiuwen--swarmflow-predictable-workflows-proof-first cd openjiuwen--swarmflow-predictable-workflows-proof-first cp .env.example .env # put your key in OPENAI_API_KEY (no quotes, no spaces), or export it instead ./prove --repeat 3 The prove script handles the entire setup for you. It creates a virtual environment (.venv) and installs the pinned packages. For each repeat, it then creates a fresh WorkSwarm home in a temp directory with jiuwenswarm-init, so your local ~/.jiuwenswarm is never touched. Next, it starts jiuwenswarm-start app. Your API key goes only into the server's process environment. The script then drives the Leader over the same WebSocket API the web UI uses (ws://127.0.0.1:19000/ws). It uses WebSocket because the headless process SDK has no Cluster mode, and SwarmFlow runs only in Cluster mode. ./prove also prints a masked form of the key it found: Using OPENAI_API_KEY= followed by the key's first 8 and last 6 characters. This allows you to check that it picked up the right key. Each repeat takes about 7 minutes (~395 seconds in the verified runs) and uses roughly 1.4M tokens, mostly cached prompt reads. The human-checkpoint scenario uses about 750k of those. Each of the three budget scenarios uses between 110k and 290k. The diagram below shows one repeat (a ./prove script run). The proof harness plays two parts. It acts as the chat user who asks the Leader to run each script. It also acts as the reviewer who answers the checkpoint. This way, a repeat runs from start to finish without waiting for you (the User). sequenceDiagram autonumber actor Dev as User participant H as Proof harness participant L as Team Leader participant SF as SwarmFlow participant E as Evidence on disk Dev->>H: ./prove H->>L: Start server, ask it to run slugify_pipeline.py L->>SF: Launch the pipeline rect rgba(76,175,80,0.10) Note over H,E: Scenario A: human checkpoint and partial re-run SF->>E: Run spec and codegen, save both to the journal SF->>H: Ask the reviewer to approve H->>SF: Reply "add max_len, cap at 40" SF->>L: Send edit points L->>SF: Edit the script, relaunch with the same run_id E->>SF: spec is replayed from the journal (CACHE_HIT) SF->>E: codegen runs again (CACHE_MISS) SF->>H: Ask the reviewer again H->>SF: Reply "approve" SF->>H: docs runs, status is completed end rect rgba(156,39,176,0.10) Note over H,E: Scenarios B, C, D: token ceilings H->>SF: Run the budget demo with no limit, a per-run limit, and a session limit SF->>H: Status is completed, failed, or stopped end H->>E: Check the journal, logs, and trace H->>Dev: Report with pass or fail for each check SwarmFlow is off by default. The test switches it on in the home's config/config.yaml with these lines from harness/setup_home.py, and you'd set the same keys in your own install: cfg["preferred_language"] = "en" # init defaults to zh when it cannot prompt cfg["setup_guide"]["enabled"] = False # first-run onboarding distracts the Leader team = cfg["modes"]["team"]["jiuwen_team"] team["enable_swarmflow"] = True # SwarmFlow is off by default team["swarmflow_budget"] = 3_000_000 # default team ceiling (chat.send may override per session) The model settings go in the home's config/.env, with no API_KEY line on disk: - MODEL_PROVIDER=OpenAI - ENDPOINT_PROFILE=openai - API_BASE=https://api.openai.com/v1 - MODEL_NAME=gpt-5.5 For platform installers and the web UI, see the WorkSwarm README and the English docs. How to define workflow stages in Python A SwarmFlow pipeline is a Python file. It imports agent, human, and phase from the swarmflow module and defines an async def run(args). The file declares the stages, and the Leader runs it with its swarmflow tool. The order of the await calls sets the order of the stages, so every run follows the same sequence. The pipeline in pipeline/slugify_pipeline.py starts with a META block. SwarmFlow reads META before running anything, so it must be a pure literal (plain values only, with no code to evaluate). Each phase title must match a phase() call further down the file. META = { "name": "slugify-pipeline", "description": "Spec -> code generation -> human review -> docs", "phases": [ {"title": "Spec"}, {"title": "Codegen"}, {"title": "Review"}, {"title": "Docs"}, ], } # The one line a reviewer's edit will change. Every agent() call is journaled under # its position in the script plus a hash of its prompt, so changing this constant # changes only the codegen call's hash (and anything downstream that embeds its output). CODEGEN_RULES = "Use only the standard library." The comment above CODEGEN_RULES describes the journal, which the next section builds on. Each agent() call is recorded under its position in the script ([["call", 0]], [["call", 1]], and so on) plus a signature hashed from its prompt. The Leader reads the script before editing it. In the proof runs, the comment also guided where the Leader made its edit. The first two stages are plain agent() calls. The label gives the call a name in the journal and the run tree. The phase argument attaches the call to a phase in META. async def run(args): # swarmflow() passes args as a string; structured input is JSON we parse ourselves. a = json.loads(args) phase("Spec") spec = await agent( f"In 3 short bullet points, specify a Python function slugify(title) for: {a['feature']}. " "Return only the bullets.", label="spec", phase="Spec", ) phase("Codegen") code = await agent( "Write ONLY the Python source (no prose, no markdown fences) implementing this spec. " f"{CODEGEN_RULES}\n\nSpec:\n{spec}", label="codegen", phase="Codegen", ) To launch the pipeline, the test sends the Leader a message that names the tool. Naming the tool makes the Leader call it directly. Run swarmflow(script_path='/tmp/swarmflow-proof-1-638pm8wc/scripts/slugify_pipeline.py', args='{"feature": "turn blog post titles into URL slugs"}'). The Leader called swarmflow with those arguments, and the tool returned run_id=wf_e0b31fee9773. This run_id names the run's journal file. You will need it again when the run resumes. How to add a human-in-the-loop checkpoint The review stage is a single await human(...) call. The prompt includes the generated code, so the reviewer can see exactly what they are approving. phase("Review") review = await human( "Review the generated code. Reply 'approve' to continue, or describe the edit you want.\n\n" + str(code) ) phase("Docs") docs = await agent( f"Write a two-sentence docstring summary for this code:\n{code}", label="docs", phase="Docs", ) return {"spec": spec, "code": code, "review": review, "docs": docs} When the run reaches human(), it pauses and the question arrives as a chat.ask_user_question event with source=swarmflow_human. The Docs phase stays planned, which means it has not started. The run tree below is built from the run's workflow.updated events. wf_e0b31fee9773 slugify-pipeline [running] run tokens=49339 budget/session : spent=179435 total=3000000 exhausted=False budget/workflow: spent=49339 total=None exhausted=False ├─ phase Spec [completed] │ └─ spec [completed] tokens=24643 ├─ phase Codegen [completed] │ └─ codegen [completed] tokens=24696 ├─ phase Review [running] │ └─ unnamed human [waiting_for_human] tokens=None ├─ phase Docs [planned] The generated slugify(title: str) had no length limit. When executed on a long title, it returned an 80-character slug: {"slug": "how-to-keep-an-ai-workflow-predictable-using-jiuwenswarm-s-swarmflow-checkpoints", "len": 80, "params": ["title"]} In the demo, the reviewer is the proof script. ./prove must give the same answer on every repeat. So harness/run_proof.py waits for the checkpoint question and sends two scripted replies over the WebSocket. The first is the edit request shown below. The second, sent later, is approve. This means ./prove never actually stops to ask you anything. When you run your own pipeline in Cluster mode, a human answers instead. As of WorkSwarm 0.2.7, the bundled web UI shows a swarmflow_human question as a question card in the chat. It sends your typed answer back in the same chat.swarmflow_reply message. The proof does not exercise this web flow, it uses the 'headless' mode instead. The proof script answered with this edit request. It was sent as chat.swarmflow_reply with the run_id and correlation_id: Edit the codegen step: slugify must take a max_len parameter (default 40) and truncate the slug to that length. Then re-run. SwarmFlow makes an extra model call to sort every human reply into one of two intents: continue or edit_rerun. There is no built-in "reject". To reject an output, have your script inspect the review string that human() returns. You can then raise an error or branch on it. This reply came back as edit_rerun, and three things followed: - The run went to paused. - The journal gained a pause record with "pause_reason": "early_return". - docs had no journal record, so it had not run. The engine then handed the Leader the edit points. Interestingly, it logged this handoff at ERROR level, as a failed tool task: ... The user asked to edit the script and re-run.\nEdit points: Modify the code generation step so slugify accepts a max_len parameter with default value 40, and truncates the generated slug to at most max_len characters. Then re-run the workflow.\n...Edit the on-disk script accordingly (do not change META.name). Edit minimally and reuse the existing script content as much as possible, refining it rather than rewriting. With no further prompt, the gpt-5.5 Leader made one edit_file call to change the script. The test never touches the script itself. Here is the change: -CODEGEN_RULES = "Use only the standard library." +CODEGEN_RULES = "Use only the standard library. Implement slugify(title, max_len=40) and truncate the generated slug to at most max_len characters." The Leader then relaunched the run with the same run_id: swarmflow({"script_path": "/tmp/swarmflow-proof-1-638pm8wc/scripts/slugify_pipeline.py", "resume_id": "wf_e0b31fee9773", "args": "{\"feature\": \"turn blog post titles into URL slugs\"}"}) On resume, SwarmFlow reloaded the journal. It compared each call's signature (a hash of its prompt) with the stored one. The spec prompt had not changed, so SwarmFlow replayed the saved result and made no model call. The codegen prompt includes CODEGEN_RULES, so its signature changed and it ran live: ... [wf] journal loaded: prior_records=4 ... 2026-10-09 15:14:17 | team | sess_a54f49d1f | WARNING | [wf] agent 'spec' key=[["call", 0]] CACHE_HIT sig=5b0917e72813 2026-10-09 15:14:17 | team | sess_a54f49d1f | WARNING | [wf] agent 'codegen' key=[["call", 1]] CACHE_MISS sig=dc49daeca408 (live run) The replayed spec text was byte-identical to the first pass. The new code returned a 40-character slug: {"slug": "how-to-keep-an-ai-workflow-predictable-u", "len": 40, "params": ["title", "max_len"]} The human() call is journaled as well. Its prompt now contained the new code, so its signature changed and the reviewer was asked again. The proof script's second reply, approve, was classified as continue. Then docs ran and the journal was sealed. Each line below shows a call's position, label, signature, and token count. The call at position 2 is the human() step, so it has no label. The last line seals the run as completed. [["call", 1]] label=codegen sig=dc49daeca408 tokens=24950 [["call", 2]] label=None sig=83f5ee8acb26 tokens=24379 [["call", 3]] label=docs sig=f0c7bc447b8f tokens=24687 {"key": "[[\"__run__\", 9223372036854775807, \"seal\", \"wf_e0b31fee9773\"]]", "type": "seal", "terminal_status": "completed"} The diagram below clearly depitcs the states a typical run can move through. stateDiagram-v2 [*] --> running: Leader calls swarmflow(script_path, args) running --> waiting_for_human: human() reached, later phases stay planned waiting_for_human --> running: reply classified continue waiting_for_human --> paused: reply classified edit_rerun (pause record early_return) paused --> running: Leader edits script, relaunches with resume_id (matching signatures replay as CACHE_HIT) running --> completed: last stage done (seal record) running --> failed: per-run workflow_token_limit exhausted (pause record) running --> stopped: team swarmflow_budget exhausted (seal record) completed --> [*] stopped --> [*] The review step also costs tokens. The unnamed human entry used 24,379 tokens, because classifying and formatting the reply each run model calls. This cost repeats every time the checkpoint asks again. How to set per-run and team token ceilings SwarmFlow enforces two ceilings: - Per-run ceiling: META["workflow_token_limit"] in the script. - Team (session) ceiling: swarmflow_budget, sent with each chat.send the same way the web UI sends it. The team ledger counts the Leader's own model calls as well as the worker agents' calls. It gates the Leader too. The demo script pipeline/budget_demo.py chains three agents and sets the per-run limit to 30,000: META = { "name": "budget-demo", "description": "outline -> draft -> headline", "phases": [{"title": "Outline"}, {"title": "Draft"}, {"title": "Headline"}], # Per-run token ceiling for this run only (the team-wide ceiling is set separately). "workflow_token_limit": 30000, } This limit looks generous for three one-sentence prompts. But each SwarmFlow worker is a full agent with a large system prompt. One agent() call cost about 25k tokens with gpt-5.5. The run crossed 30,000 during draft, so headline never started. For the budget scenarios, the Leader prompt adds "Do not edit the script and do not relaunch it yourself." The diagnostic tells the Leader to redesign the script and relaunch. The demo needs to show the stopped run as it is, so the Leader must not do that. wf_826df91268e1 budget-demo [failed] run tokens=49258 budget/session : spent=222723 total=3000000 exhausted=False budget/workflow: spent=49258 total=30000 exhausted=True budget_exhausted_scope=workflow ... ├─ phase Headline [failed] │ └─ headline [failed] tokens=None The diagnostic reaches the Leader as a follow-up message. The test reads it from the home's .jiuwenswarm/.trace/ store. It names the layer that tripped, the heaviest phases, and the recovery path: [Swarmflow budget exhausted] run_id=wf_826df91268e1 workflow token budget exhausted: 49258/30000 Triggered layer: workflow (per-run) (spent=49258/30000). Heaviest phases: Draft(24653), Outline(24605). This ceiling is the run's per-run token budget (independent of the session total), set by the user and must NOT be changed; you must redesign the workflow to consume fewer tokens (simplify / fewer agents / cheaper model) and relaunch. Relaunch by passing resume_id=<the run_id above> together with the edited script_path: ... The run ends as failed with a workflow_budget_exhausted pause record, so it can still be resumed. The guidance warns that replayed calls are "re-billed from their stored tokens". In other words, a resume counts them against the same ceiling again. The proof did not test a resume after exhaustion. For the team ceiling, the test ran budget_demo_unlimited.py, which has no per-run limit. It set swarmflow_budget to the Leader's pre-launch spend plus one worker call. Both values were measured in an unconstrained control run. In repeat 1, that came to 43,218 + 24,643 = 67,861. A fixed small number would have stopped the Leader before it launched anything. wf_df24a8c15beb budget-demo-unlimited [stopped] run tokens=24607 budget/session : spent=111143 total=67861 exhausted=True budget/workflow: spent=24607 total=None exhausted=False budget_exhausted_scope=session ... Triggered layer: session (shared) (spent=111143/67861). Heaviest phases: Outline(24607). This is the team's shared (session-wide) token ceiling and is currently exhausted. Until the ceiling is raised, relaunching or starting a new workflow will hit the same gate immediately — raise the ceiling and open a new session before retrying swarmflow. This time, the journal got a seal record with "terminal_status": "stopped". The Leader is gated by the same ledger, so it sent no final message. Use the per-run limit to cap a single script you expect to iterate on. Set swarmflow_budget as the hard spend cap for the whole team session. How to monitor a run from its workflow.updated events WorkSwarm also has an optional terminal UI, jiuwenswarm-tui. This tutorial monitored runs from the event stream that the UIs read. That approach works wherever WorkSwarm runs. The run trees shown earlier come from the workflow.updated events on the web channel WebSocket. Each event carries: - The run status. - The token_count. - Both budget ledgers (budget and workflow_budget). - The phases, with per-agent status and tokens. These events are deltas. Each one carries only the phases that changed. A monitor therefore has to merge them into one snapshot, as merged() does in harness/run_proof.py: top, phases = {}, {} for f in frames: if not (is_sess(f, sid) and etype(f) == "workflow.updated" and wf_of(f).get("id") == run_id): continue wf = wf_of(f) top.update({k: v for k, v in wf.items() if k != "phases"}) for ph in wf.get("phases") or []: cur = phases.setdefault(ph.get("name"), {"agents": {}}) cur.update({k: v for k, v in ph.items() if k != "agents"}) for a in ph.get("agents") or []: cur["agents"].setdefault(a.get("id"), {}).update(a) Across the four scenarios, the merged trees showed these phase and agent statuses: planned, running, waiting_for_human, paused, completed, failed and stopped. The swarmflow tool also supports run control. You pass resume_id together with an action, and the tool routes pause, resume and stop requests to the background-task controller (tool_swarmflow.py::_control_run). The proof does not exercise this, so test it before you rely on it. To see the events for yourself, run ./prove --repeat 1 and open proof-out/repeat-1/transcript.md. Confirm that the unnamed human [waiting_for_human] entry sits under phase Review, while phase Docs is still planned. What to build next with SwarmFlow You now have a working SwarmFlow setup with three concrete capabilities: - Fixed stage order declared in a Python file, so every run follows the same path regardless of what the Leader decides at runtime. - A human checkpoint that pauses the run, accepts an edit request, and resumes under the same run_id with unchanged stages replayed from the journal. - Two token ceilings that each stop a run with layer-specific advice the Leader can act on immediately. Swap the slugify stages for real stages in your pipeline, put human() right before the most expensive (or crucial) step, and size workflow_token_limit from the per-agent token counts in your own run tree. Star the WorkSwarm repo, follow the openJiuwen-ai organization for releases, and check the English docs for the full configuration reference. Contributions and issue reports are welcome. --- Frequently asked questions What is SwarmFlow in WorkSwarm? SwarmFlow is WorkSwarm's scripting layer for Cluster mode (team mode) that fixes the execution order of a multi-agent pipeline in a Python file. Each stage is an await agent() or await human() call, and the Leader runs the file with its swarmflow tool, so the stage sequence is identical on every run regardless of runtime Leader decisions. How does the SwarmFlow journal prevent redundant LLM calls after an edit? Each agent() call is recorded under its position in the script and a signature hashed from its prompt. On resume, SwarmFlow compares each call's current signature against the stored one. Calls whose prompts are unchanged replay from the journal as CACHE_HIT without hitting the model, while calls whose prompts changed run live. In the proof run, editing CODEGEN_RULES produced a CACHE_HIT for spec and a CACHE_MISS for codegen, cutting the resume cost to a single live agent call. What is the difference between the per-run token ceiling and the team token ceiling? The per-run ceiling is META["workflow_token_limit"] in the pipeline script and caps token spend for that single run. The team ceiling is swarmflow_budget, set per chat.send, and counts every model call in the session including the Leader's own calls. Exhausting the per-run ceiling ends the run as failed with a resumable pause record. Exhausting the team ceiling ends the run as stopped and gates the Leader from launching anything new until the budget is raised. Why doesn't ./prove ask me for input at the human checkpoint? The proof script answers the human checkpoint for you. To prove the same behavior on every repeat, harness/run_proof.py waits for the chat.ask_user_question event and sends two scripted replies through chat.swarmflow_reply, the edit request and then approve. The checkpoint still blocks the run, and docs doesn't start until a reply arrives. In your own Cluster-mode session, the WorkSwarm web UI shows the question as a card in the chat, and you can type your reply there. How does SwarmFlow classify a human reply as an edit request vs. an approval? SwarmFlow runs an extra model call on every human() reply to classify it as either continue or edit_rerun. A reply of approve routes to continue and the pipeline resumes from the next stage. Any reply describing a change routes to edit_rerun, which pauses the run, writes a pause record with "pause_reason": "early_return", and passes structured edit points to the Leader so it can modify the script and relaunch with resume_id. What happens to a run that hits the team token ceiling mid-stage? The run ends as stopped with a seal record instead of a resumable pause record. The Leader, gated by the same ledger, sends no final message. The diagnostic names the session layer, the heaviest phase, and the instruction to raise the ceiling and open a new session before retrying swarmflow. How do you prevent the Leader from automatically relaunching a budget-exhausted run? Add the instruction "Do not edit the script and do not relaunch it yourself." to the Leader prompt before triggering the budget scenario. Without it, the diagnostic message tells the Leader to redesign and relaunch, and gpt-5.5 will act on that instruction immediately, which obscures the stopped-run state you may want to inspect. ---

模型连接测试通过，Agent 就能干活了吗？ JiuwenSwarm 的 develop 文档说明：Test 只发简单请求；默认模型还需支持工具调用和多轮对话。 建议再验收：确认当前模型和工具权限，在测试目录完成“读文件→改副本→检查结果”，记录失败步骤。 按安装版本核对配置，再接真实项目。 github.com/openJiuwen-ai/jiuwenswarm/blob/develop…

Debugging a silently failing AI workflow with Agent Core full-link observability The most difficult AI workflows to debug are the ones that don't crash. A multi-step agent pipeline can quietly return a plausible-looking answer that's actually wrong, without ever throwing an error. One broken step early in the chain compounds into a wrong final output, and standard monitoring rarely catches it: the run finishes, the logs show green, and the problem only surfaces hours later when a downstream team flags that something's off. This tutorial walks through a concrete case using openJiuwen Agent Core v0.1.18. You'll instrument a three-node pipeline (fetch → analyze → summarize) with OTEL-compatible spans (structured traces that record what each step did, in a format most observability tools understand). Then you will read back their output (latency, token counts, and a per-step failure flag), and pinpoint the exact node where the chain went wrong. The complete companion code is available at the tutorial repository. Why silent failures escape standard logs A silent failure happens when a step in an agent pipeline returns something that looks like a normal result but isn't. No exception is raised, no log line turns red, and no HTTP error code shows up, even though the output is wrong. Standard Python logging and HTTP status codes are built to catch exceptions. Agent-specific failures often don't rise to that level, so they slip straight past. Three failure modes show up most often in production pipelines: 1. Incorrect tool selection: The agent picks the wrong function for the job. That function still runs, still returns, and still fills the Observation (the agent's record of what a tool call returned) with data, just data that has nothing to do with the task. No exception fires, because nothing actually went wrong from the tool's point of view. 2. Hallucinated tool arguments: The agent calls the right tool, but with parameters it made up or left out. If the tool tolerates missing input rather than rejecting it, the call succeeds and returns something empty or nonsensical. 3. Context truncation: The prompt is too long for the model's context window (the amount of text it can consider at once), and the model API returns a normal HTTP 200 response with an empty list of completions. The token count comes back as zero, no error is raised, and the node exits as if nothing happened. This matters more in a ReAct-style agent loop, which cycles through Thought, Action, and Observation, because each step can look successful on its own while the whole chain quietly drifts off course. A truncated completion still produces something that reads as a valid string, so the next Thought step picks it up and keeps going as if nothing were wrong. By the time the loop finishes, the final output is wrong, but there's still no exception and no red log line to explain why. Sample workflow setup You'll need Python 3.11, 3.12, or 3.13 (openjiuwen doesn't yet support 3.14). Clone the repository, change into the project directory, and install the dependencies: git clone https://github.com/manishh/openjiuwen-debugging-an-ai-workflow-that-fails-silently-with-agent cd openjiuwen-debugging-an-ai-workflow-that-fails-silently-with-agent pip install -r requirements.txt See the openJiuwen quickstart documentation for initial API key setup. The tutorial reads five environment variables: OPENJIUWEN_API_KEY, OPENJIUWEN_MODEL, OPENJIUWEN_API_BASE, OPENJIUWEN_PROVIDER, and OPENJIUWEN_SSL_VERIFY. If you leave out OPENJIUWEN_API_KEY, the code runs in offline simulation mode, which still emits and analyzes spans and only skips the live model call. The tutorial's target pipeline is a sentiment-analysis workflow with three connected steps: - fetch_data retrieves and preprocesses the source text. - analyze_sentiment classifies its sentiment. This is the node that simulates a silent failure. - summarize generates a human-readable summary. The Thought/Action/Observation loop above describes how a live ReActAgent reasons. The pipeline in src/main.py simulates the same failure pattern with three plain Python functions, so it runs deterministically offline without a live agent loop. The agent construction in examples/02_workflow_agent_setup.py looks like this: card = AgentCard( id="sentiment-analysis-pipeline", name="sentiment-analysis-pipeline", description="Fetch -> analyse -> summarise sentiment pipeline", ) config = ReActAgentConfig().configure_model_client( provider=PROVIDER, api_key=API_KEY, api_base=API_BASE, model_name=MODEL, verify_ssl=SSL_VERIFY, ) # The agent below drives the three-node pipeline: # node_fetch_data -> node_analyze_sentiment -> node_summarize agent = ReActAgent(card=card).configure(config) print(f"Agent '{agent.card.name}' created.") ReActAgent is openjiuwen's current general-purpose agent, built around the Thought/Action/Observation loop described earlier. It decides at runtime which tool to call next, rather than following a hardcoded step order. The older WorkflowAgent, which ran a fixed, user-defined node sequence, is deprecated as of openjiuwen 0.1.18 in favor of this AgentCard + ReActAgentConfig pattern. This tutorial's pipeline order (fetch → analyze → summarize) is enforced by the plain Python function calls in src/main.py, not by the agent, so the fixed order holds regardless of which agent class builds it. Full-link observability in Agent Core Full-link observability in Agent Core means that every tool call, token count, and step latency is captured as a structured, OTEL-compatible span (a self-contained record of one unit of work). Chain those spans together and you get a complete execution trace that you can query programmatically or view in Agent Studio, with no third-party APM (Application Performance Monitoring) tool required. As of v0.1.18, Agent Core emits this telemetry natively. To capture and inspect those traces locally, you attach exporters to a TracerProvider instance. The full setup is in examples/01_setup_observability.py: from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor, ConsoleSpanExporter from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter exporter = InMemorySpanExporter() provider = TracerProvider() provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter())) provider.add_span_processor(SimpleSpanProcessor(exporter)) # Always get the tracer from your provider, not from opentelemetry.trace.get_tracer(). tracer = provider.get_tracer("agent-core.tutorial") SimpleSpanProcessor exports each span the moment it finishes, so the console output and the trace report you'll build later stay in the order the spans actually ran. Get your tracer from your own provider instance (provider.get_tracer(...)) rather than from the global opentelemetry.trace module. If Agent Core has already initialized the global provider, calling trace.get_tracer() returns a tracer bound to the SDK's own exporter, and your spans go somewhere you can't see them. With the tracer in hand, wrap every step in a custom span that records four diagnostic signals. The broken analyze_sentiment node in examples/03_custom_spans.py shows what a silently failing node looks like once it’s instrumented: def node_analyze_sentiment(data: dict) -> dict: with tracer.start_as_current_span("tool.analyze_sentiment") as span: t0 = time.perf_counter() # Simulate: context window overflow caused the model to return # an empty choices list. Status code was 200 – no error logged. time.sleep(0.15) # slightly elevated latency: the client waited for timeout result = {"sentiment": "", "token_count": 0, "confidence": 0.0} span.set_attribute("agent.tool.name", "analyze_sentiment") span.set_attribute("agent.tool.token_count", result["token_count"]) span.set_attribute("agent.step.latency_ms", round((time.perf_counter() - t0) * 1000, 2)) # This flag is the key signal in the trace: token_count==0 is a silent failure. span.set_attribute("agent.step.silent_failure", result["token_count"] == 0) return result Run the script (python examples/03_custom_spans.py), and the span summary makes the failure obvious: Span summary: tool.fetch_data tokens= 312 latency= 40.15 ms silent_failure=False tool.analyze_sentiment tokens= 0 latency= 150.15 ms silent_failure=True tool.summarize tokens= 144 latency= 50.13 ms silent_failure=False Look at the analyze_sentiment row: tokens=0 and silent_failure=True, even though the node returned normally with no exception raised. Your latency numbers will vary slightly between runs, since they come from a real time.sleep() call, but that row is the one to watch either way. The last line of the code snippet is the one that matters most. Every node sets agent.step.silent_failure, not just the broken one, and it’s True whenever token_count == 0. That's what turns "did this succeed?" into something you can actually query. Together, the four attributes (agent.tool.name, agent.tool.token_count, agent.step.latency_ms, and agent.step.silent_failure) are the signals you'll read in the next section. Because every node writes all four, the trace report has a consistent shape to scan no matter which node broke. How to read the trace output The diagram below shows the full-link trace for a broken run. The root workflow.session span wraps all three tool spans, and tool.analyze_sentiment is highlighted because its agent.step.silent_failure attribute is True. flowchart LR A["workflow.session<br/>root span"] --> B["tool.fetch_data<br/>tokens: 312<br/>latency: 40 ms<br/>silent_failure: False"] B --> C["tool.analyze_sentiment<br/>tokens: 0<br/>latency: 150 ms<br/>silent_failure: True"] C --> D["tool.summarize<br/>tokens: 144<br/>latency: 50 ms<br/>silent_failure: False"] style C fill:#ff6b6b,color:#fff Once the workflow finishes, iterate over exporter.get_finished_spans() to read each span back; each one maps to a single node. The reporting loop in examples/04_read_trace_output.py prints the broken and fixed runs one after the other: def _report(label: str, spans: list) -> None: print(f"\n{'='*72}\n{label}\n{'='*72}") print(f"{'Span':<35} {'tokens':>7} {'latency ms':>11} {'failure':>8}") print("-" * 72) for s in spans: a = s.attributes or {} flag = "◄ FAIL" if a.get("agent.step.silent_failure") else "OK" print( f"{s.name:<35} {str(a.get('agent.tool.token_count','-')):>7} " f"{str(a.get('agent.step.latency_ms','-')):>11} {flag:>8}" ) Running it prints both reports, so you can compare them line by line: ======================================================================== BROKEN RUN (inject_failure=True) ======================================================================== Span tokens latency ms failure ------------------------------------------------------------------------ tool.fetch_data 312 40.0 OK tool.analyze_sentiment 0 150.0 ◄ FAIL tool.summarize 144 50.0 OK workflow.session - - OK ======================================================================== FIXED RUN (inject_failure=False) ======================================================================== Span tokens latency ms failure ------------------------------------------------------------------------ tool.fetch_data 312 40.0 OK tool.analyze_sentiment 87 40.0 OK tool.summarize 144 50.0 OK workflow.session - - OK The comparison shows the pattern to look for: - A healthy node shows token_count > 0, latency_ms within the expected range for that step, and OK. - A broken node shows token_count == 0 because the model returned nothing, elevated latency_ms, and ◄ FAIL. In this simulation, the latency is elevated because the client waited out a timeout before accepting the empty response. Running src/main.py produces the same pattern from a single script, without the side-by-side comparison. Its analyse_trace() function reads from the InMemorySpanExporter and prints: ⚠ First broken span: tool.analyze_sentiment token_count = 0 latency_ms = 150.24 workflow.node= node_analyze_sentiment In Agent Studio's online debugging environment, the same execution graph appears visually in the session view, with the broken span flagged in the error log alongside its attributes. Because Agent Studio reads the same structured trace data Agent Core emits, you don't need to run analyse_trace() by hand once you're working against a live session there. Each failure mode leaves its own signature in the spans: | Failure mode | Span signature | |---|---| | Context truncation | token_count == 0 with elevated latency | | Hallucinated arguments (no validation) | token_count > 0, but the output is semantically empty or causes a schema error in a downstream span; latency is usually normal | | Hallucinated arguments (with the validation guard below) | token_count == 0, normal latency, and an agent.tool.arg_error attribute naming the bad argument | | Incorrect tool selection | An agent.tool.name that doesn't match the expected step, with output that doesn't fit the task | Three targeted fixes from span evidence You now have the failing span in front of you. The next step is turning that evidence into a fix. Each of the three failure modes leaves a distinct span signature, and each signature points to one specific code change. Context truncation (token_count == 0, elevated latency) Add a token budget check before submitting the prompt. The fix in examples/05_apply_fixes.py guards the analysis node: MAX_PROMPT_TOKENS = 3800 # leave headroom for the model's completion budget def node_analyze_sentiment_fixed(data: dict[str, Any]) -> dict[str, Any]: with tracer.start_as_current_span("tool.analyze_sentiment") as span: t0 = time.perf_counter() prompt_text = data.get("text", "") estimated_tokens = _token_estimate(prompt_text) if estimated_tokens > MAX_PROMPT_TOKENS: # Truncate to fit. In production use a summarisation sub-call instead. chars_allowed = MAX_PROMPT_TOKENS * 4 prompt_text = prompt_text[:chars_allowed] span.set_attribute("agent.prompt.truncated", True) # Simulate the model call with a valid response after truncation. time.sleep(0.04) result = {"sentiment": "positive", "token_count": 87, "confidence": 0.91} ... The * 4 uses the rough rule of thumb that one token is about four characters of English text. In production, replace _token_estimate with a real tokenizer (tiktoken, for example). The span's agent.prompt.truncated attribute records that truncation happened, so you can monitor for prompts that keep hitting the ceiling. Hallucinated tool arguments (token_count > 0 but output semantically empty, or argument schema errors in downstream spans) Validate arguments at the tool boundary and record any error in the span. The invoke_tool_with_validation function in examples/05_apply_fixes.py shows the pattern: def invoke_tool_with_validation( tool_name: str, args: dict[str, Any] ) -> dict[str, Any]: with tracer.start_as_current_span(f"tool.{tool_name}") as span: span.set_attribute("agent.tool.name", tool_name) valid, error = _validate_tool_args(tool_name, args) if not valid: # Reflect the validation error back into the span so the ReAct # Observation step can self-correct on the next iteration. span.set_attribute("agent.tool.arg_error", error) span.set_attribute("agent.step.silent_failure", True) span.set_attribute("agent.tool.token_count", 0) return {"error": error, "token_count": 0} ... Running the script exercises both a valid call and a rejected one, so you can see the guard working on real span output: --- Fix 2: Argument validation (valid args) --- result={'output': 'Result from analyze_sentiment', 'token_count': 95} --- Fix 2: Argument validation (hallucinated / missing args) --- result={'error': "Missing required args for analyze_sentiment: ['text']", 'token_count': 0} The rejected call still produces token_count == 0, the same signal as context truncation. The difference is that it's caught before any model call happens, its latency stays normal, and the returned error names the missing argument. That's how you tell the two failure modes apart in a real trace: check the tool's result payload and the agent.tool.arg_error attribute, not just the token count. Returning the structured error in the Observation also lets a ReActAgent loop self-correct on the next iteration rather than continuing with a bad result. Incorrect tool selection (wrong agent.tool.name in the span, output mismatches intent) Here the span shows a tool name that doesn't match the expected step. The fix is on the prompt side, so the agent picks the correct function on the first attempt: - Tighten the tool description schema. - Add a worked example of the correct tool call. - Constrain the output format, so the agent has less room to substitute a plausible-sounding alternative. This fix changes the prompt rather than the tool code, so it has no span-level output of its own to show, and examples/05_apply_fixes.py doesn't demonstrate it. Verifying the pattern holds The smoke tests in tests/smoke_test.py check that the underlying span mechanics work: that a span with token_count == 0 is picked up as silent_failure == True, and that a multi-span workflow isolates the one broken node from the healthy ones. def test_otel_silent_failure_detection(): ... broken = [s for s in finished if s.attributes.get("agent.step.silent_failure")] assert len(broken) == 1, "Expected one silent-failure span" assert broken[0].attributes["agent.tool.token_count"] == 0 These tests build their own spans directly rather than importing from examples/05_apply_fixes.py. They confirm that the detection pattern itself is sound, not that any specific fix in this section works. After changing a fix, rerun the example script and compare its span report against the output shown above, then run pytest tests/smoke_test.py to confirm the detection logic underneath still works. The output of smoke_test.py should show PASSED for five tests. collected 5 items tests/smoke_test.py::test_openjiuwen_sdk_imports PASSED [ 20%] tests/smoke_test.py::test_otel_span_creation PASSED [ 40%] tests/smoke_test.py::test_otel_silent_failure_detection PASSED [ 60%] tests/smoke_test.py::test_otel_multi_span_workflow PASSED [ 80%] tests/smoke_test.py::test_openjiuwen_agent_construction PASSED [100%] Note that run_workflow(inject_failure=False) in src/main.py shows what a healthy run looks like after a fix: it returns the same token_count=87 result as the guard above, but it doesn't contain the fix logic itself. The guard, the validation, and the prompt-tightening code live in examples/05_apply_fixes.py; main.py only toggles between the broken and fixed outcomes to show the contrast in the trace report. Closing thoughts The debugging loop here is repeatable, and it's worth building into a habit rather than treating as a one-off fix: 1. Instrument every tool node with the same four span attributes: agent.tool.name, agent.tool.token_count, agent.step.latency_ms, and agent.step.silent_failure. 2. Run the workflow and read back the finished spans. 3. Scan for zero-token observations and latency spikes, and the broken node names itself. The same idea extends to multi-agent collaboration. Agent Core v0.1.18 added ReliabilityRail to the agent_teams package, which you can attach to a team of agents. Its detectors watch for repeated tool calls, model errors, tool errors, output-length problems, context-compaction issues, and agents bouncing work back and forth. When one trips, it can correct the problem locally, escalate to the team Leader, or notify you. Once you're running several agents in one session, those signals roll up into a per-session error rate, automating much of what you just did by hand. To explore the SDK and start tracing your own agents, see the openJiuwen documentation. To see full-link observability and Agent Studio in action, contact the openJiuwen team for a guided walkthrough. --- Frequently asked questions (FAQs) How do I know which node caused a wrong output when no exception was raised? Scan your finished spans for agent.step.silent_failure: True combined with agent.tool.token_count == 0. That combination identifies the exact node where the model returned nothing. Elevated agent.step.latency_ms alongside a zero token count narrows the cause to context truncation specifically. What's the difference between a context truncation failure and a hallucinated argument failure in the span data? Context truncation produces token_count == 0 with elevated latency. Without argument validation, a hallucinated argument failure typically shows token_count > 0 (the model did respond), but the output is semantically empty or triggers a schema error in a downstream span, usually with normal latency. With the validation guard in place, the bad call is rejected before the model runs, so it shows token_count == 0 with normal latency and an agent.tool.arg_error attribute that names the problem. Do I need Agent Studio to use full-link observability, or does it work locally? Full-link observability works entirely locally. Attach an InMemorySpanExporter and a ConsoleSpanExporter to your TracerProvider, run the workflow, call provider.force_flush(), and iterate over exporter.get_finished_spans(). Agent Studio reads the same structured trace data and shows it visually, but emitting the spans requires no external service. Why should I get the tracer from my own provider instance rather than the global opentelemetry.trace module? If Agent Core has already initialized the global provider, calling trace.get_tracer() returns a tracer bound to the SDK's exporter. Your custom spans then go to the wrong destination and won't appear in your InMemorySpanExporter. Always call provider.get_tracer(...) on your own TracerProvider instance. Can I apply the same four-attribute span pattern to a multi-agent team, not just a single pipeline? Yes. The openjiuwen.agent_teams package lets you attach ReliabilityRail (available in v0.1.18+) to a team of agents. It adds configurable detection for repeated tool calls, model errors, tool errors, output-length anomalies, context compaction, and agents bouncing work back and forth. In production, the per-session error rates from Agent Core's structured traces become the signal you alert on across the whole team. What should I do if prompts consistently hit the token ceiling after I add the budget check? Replace the truncation approach with a summarization sub-call that condenses the input before it reaches the analysis node. The agent.prompt.truncated span attribute tells you how often truncation fires, so you can set a threshold and trigger the sub-call automatically when the attribute appears more than a configurable number of times per session. How does the ReActAgent loop self-correct after a hallucinated argument error? When invoke_tool_with_validation detects an invalid argument, it returns a structured {"error": ..., "token_count": 0} dict into the Observation. The ReActAgent reads that Observation on the next Thought step and can revise its tool call with corrected arguments, instead of continuing from a bad result with no signal that anything went wrong. ---

How to get multiple AI agents working together on one task: a WorkSwarm walkthrough Most multi-agent demos fall apart at the handoff. One agent returns a paragraph when the next one expects a list. A subtask stalls, and nothing notices. Before long, you are writing dispatch loops and retry logic around every model call. This tutorial shows a better way. You will build a Cluster-mode Swarm in WorkSwarm (formerly JiuwenSwarm), where a Leader hands a product-analysis task to three specialist Teammates. Every step's output is checked against a JSON schema (an agreed output format), so the next agent always gets the shape it expects. You can even test the whole pipeline offline with pytest before it touches a paid model. Along the way, you will - write the SwarmFlow script that defines the pipeline - configure the models your Teammates use - run the smoke tests offline - watch a live run from the Web UI The complete code is in the companion repository. Prerequisites and installation You will need: - Python 3.11, 3.12, or 3.13 - API credentials for at least one supported LLM provider WorkSwarm is published on PyPI as workswarm, and installing it also pulls in Agent Core and the SwarmFlow engine. The companion repository pins version 0.2.6 in requirements.txt. Clone it and install its dependencies: git clone https://github.com/manishh/openjiuwen-how-to-get-multiple-ai-agents-actually-working-together cd openjiuwen-how-to-get-multiple-ai-agents-actually-working-together pip install -r requirements.txt Next, copy the companion repository's example config into the WorkSwarm config directory: mkdir -p ~/.jiuwenswarm/config/ cp config/config.yaml ~/.jiuwenswarm/config/config.yaml Then initialize the workspace and start the services. The command-line tools kept their original names after the rename, so they still start with jiuwenswarm: jiuwenswarm-init jiuwenswarm-start jiuwenswarm-init creates the ~/.jiuwenswarm/ workspace on first run. Because you’ve already copied config.yaml into that directory, init merges it with the default version, so none of your settings are lost. jiuwenswarm-start launches the Gateway, the AgentServer, and the Web UI at http://localhost:5173: !WorkSwarm If you'd rather skip pip, the install guide has one-click desktop installers for Windows and macOS, and the README also lists a HarmonyOS desktop version. How WorkSwarm's coordination engineering model works WorkSwarm splits a multi-agent job between a Leader and a team of Teammates. The Leader reads the goal, forms the team, and breaks the work into tasks. Each Teammate claims a task that fits its role. If a Teammate gets blocked, it escalates to the Leader (asks the Leader to step in). Once the work is done, the Leader merges the results into the final deliverable. openJiuwen calls this approach coordination engineering. The Agent Team guide walks through the full Leader and Teammate cycle. By default, the Leader plans that cycle adaptively, choosing the order as it goes. That flexibility suits exploratory work, but two runs can take different paths. When you need a repeatable pipeline, write a SwarmFlow script. You define the stages as a Python file, and the Leader runs them in the order your code sets. The pipeline keeps the same shape every time, and it lives in version control, where you can review and test it. Failure handling lives in Agent Core, the engine underneath WorkSwarm. The Agent Core v0.1.18 release added ReliabilityRail, a set of six detectors that watch for repeated tool calls, model errors, tool errors, output-length problems and context-compaction issues, and that catch agents bouncing work back and forth. Each detector has configurable severity actions: correct the problem locally, escalate to the Leader, or notify the user. This is a separate path from Teammate escalation: a Teammate escalates when it is blocked, while ReliabilityRail steps in when it detects a failure pattern. The same release also made third-party agents more resilient: transient timeouts, rate limiting (429), and service-unavailable (5xx) errors are retried, with guidance surfaced to the user and consistent final error codes. The diagram below shows how this maps to the pipeline you will build. The Leader runs the script through three phases: research, summary, and report. ReliabilityRail watches the run and corrects, escalates, or notifies when something goes wrong. flowchart TD L["Leader<br/>runs swarmflow_pipeline.py"] --> R["Phase 1: Research<br/>market_researcher"] R -->|"trends, competitors, opportunities"| S["Phase 2: Summarize<br/>data_summarizer"] S -->|"exactly 5 bullets"| P["Phase 3: Report<br/>report_writer"] R -->|"full research"| P P -->|"report"| L E["Agent Core<br/>ReliabilityRail"] -.->|"correct/escalate/notify"| L Leader agent configuration and task decomposition A SwarmFlow script has two required top-level pieces: a literal META dict that describes the pipeline, and an async def run(args) entry point that the engine awaits. Everything inside run() is ordinary Python plus three operators the engine supplies: agent() to call a Teammate, phase() to mark a stage, and log() to write a progress line. import json from swarmflow import agent, phase, log META: descriptive, and it must be a literal The engine reads META from the file's syntax tree without importing the script, so it has to be a plain dict literal. A META built by a function call or computed value is rejected with a MetaError. That's also why the engine can list your pipeline before running any of its code. META: dict = { # Human-readable identifier surfaced in the Web UI (http://localhost:5173) "name": "product-analysis-pipeline", # Shown on the Swarm task card "description": ( "Three-stage product-analysis Swarm: market research → " "data summarization → executive report. Demonstrates Cluster-mode " "Leader/Teammate coordination." ), "version": "1.0.0", # Subtask metadata used for documentation and the Web UI task-progress panel. # The engine reads the actual dependency graph from the pipeline # structure below, not from this list. "subtasks": [ { "role": "market_researcher", "description": "Gather and structure market data", "success_criteria": "JSON with keys: trends, competitors, opportunities", }, { "role": "data_summarizer", "description": "Distill the research into five actionable bullets", "success_criteria": "JSON with a 'bullets' list of exactly 5 strings", }, { "role": "report_writer", "description": "Write the executive report from research and summary", "success_criteria": "JSON with Executive Summary, Key Findings, Recommendations", }, ], } Treat subtasks as documentation. Nothing in it schedules or enforces anything. The real dependency graph is the order of the awaits in run(), and the real contract is the schema attached to each call. Schemas make each step's contract enforceable The success_criteria string says what a step should return. The schema= argument is what the engine checks. Each step returns a parsed, validated object that the next step can use directly, with no string parsing in between. This is a new feature introduced in the latest WorkSwarm for obtaining correct and structured output for agents to work with. RESEARCH_SCHEMA = { "type": "object", "properties": { "trends": {"type": "array", "items": {"type": "string"}}, "competitors": {"type": "array", "items": {"type": "string"}}, "opportunities": {"type": "array", "items": {"type": "string"}}, }, "required": ["trends", "competitors", "opportunities"], } SUMMARY_SCHEMA = { "type": "object", "properties": { "bullets": { "type": "array", "items": {"type": "string"}, "minItems": 5, "maxItems": 5, }, }, "required": ["bullets"], } REPORT_SCHEMA = { "type": "object", "properties": { "executive_summary": {"type": "string"}, "key_findings": {"type": "array", "items": {"type": "string"}}, "recommendations": {"type": "array", "items": {"type": "string"}}, }, "required": ["executive_summary", "key_findings", "recommendations"], } minItems and maxItems are what turn "exactly 5 bullets" from a request in the prompt into a rule the engine checks. run(): plain awaits, each feeding the next The three steps (market research, data summarization, and report writing) depend on each other, so they are sequential awaits. The body of run() is a chain of three agent() calls. Each call sends a prompt to a Teammate, waits for the reply, and returns it as a parsed object that already matches its schema. The next call embeds that object in its own prompt. async def run(args): task = args["task"] # Phase 1: research phase("Research") log("[Leader] Phase 1 - dispatching market_researcher Teammate") research = await agent( f"{task}\nReturn the trends, competitors and opportunities you find. " "Keep it short and return results quickly.", label="market_researcher", schema=RESEARCH_SCHEMA, ) The first step is the only one that sees the raw task. Its schema guarantees research is a dict with trends, competitors and opportunities, each a list of strings. The instruction to keep the answer short and fast keeps the walkthrough quick. Drop it if you want deeper research (it will take longer). # Phase 2: summarize phase("Summarize") log("[Leader] Phase 2 - dispatching data_summarizer Teammate") summary = await agent( "Distill this market research into exactly 5 concise, actionable bullet points:\n" + json.dumps(research), label="data_summarizer", schema=SUMMARY_SCHEMA, ) The summarizer never sees the original task. It receives only research, serialized with json.dumps so it can be embedded in the prompt as text. Because research was validated in step 1, this step can rely on its shape. SUMMARY_SCHEMA enforces the "exactly 5" requirement, which is why the result is an object with a bullets list. # Phase 3: report phase("Report") log("[Leader] Phase 3 - dispatching report_writer Teammate") report = await agent( "Write an executive report with sections 'Executive Summary', 'Key Findings' " "and 'Recommendations' from this research and summary:\n" + json.dumps({"research": research, "summary": summary}), label="report_writer", schema=REPORT_SCHEMA, ) return {"report": report} The writer gets both earlier outputs, the full research and the five-bullet summary, so it can use the detail from step 1 and the priorities from step 2. This is the one place where data fans in: step 3 reads from two earlier steps, not just the one before it. REPORT_SCHEMA requires the three named sections, and run() returns the validated report as {"report": report}. Three details about how this behaves: - Order comes from the code. Each await blocks until its Teammate finishes, so step 2 can't start before step 1 returns. Nothing else defines the dependency graph, and META["subtasks"] doesn't affect it. - phase(), log() and label= are for readability. They group and name steps in the run's progress view and event stream. Removing them wouldn't change what the pipeline computes. - Completed steps are journaled. If a run is resumed from its journal, finished steps replay from the record instead of calling the model again, so a failure in step 3 doesn't force you to pay for steps 1 and 2 twice. Testing without a model The swarmflow module, which provides agent, log and phase, only exists while the engine is executing a script. Running python src/swarmflow_pipeline.py directly therefore fails with an import error. Instead, the repository's tests drive the script through Agent Core's run_workflow() with its built-in MockBackend, which returns deterministic replies that match each schema. All nine tests run without an API key. pytest -v tests/smoke_test.py The tests check that: - the agent, log and phase operators resolve - META and run() have the right shape - all three agent() calls start and complete - a resumed run replays from its journal without calling the model again They also catch API misuse early. For example, passing an unsupported keyword such as role= to agent() fails here with a TypeError instead of later in the Web UI. Choosing a model per step The label="market_researcher" on an agent() call is only a name that shows up in the run tree; it doesn't pick a model. Every agent() call runs on the teammate model by default. To run a step on a different model, name that model on the call: research = await agent( prompt, label="market_researcher", schema=RESEARCH_SCHEMA, model="model_name", ) This keeps the pipeline portable: provider details and credentials stay in WorkSwarm's configuration, and the script only says which model a step should use. Setting up providers You can add providers in the Web UI (Settings > Models) or in ~/.jiuwenswarm/config/config.yaml. Each model entry has: - model_name, for example gpt-4o or deepseek-chat - api_base, up to the version level only (for example https://api.openai.com/v1); WorkSwarm appends /chat/completions itself - api_key - client_provider - alias (optional; defaults to model_name), which is the name you refer to the model by One entry is marked with is_default: true, and that's the fallback when nothing else is specified. WorkSwarm supports OpenAI, DeepSeek, DashScope, SiliconFlow, InferenceAffinity, OpenRouter and Huawei Cloud MaaS, plus any OpenAI-compatible endpoint. The configuration guide has the full reference. !WorkSwarm: Configure Model You can also supply keys as environment variables, which take precedence over values in the config file. Export them before you start the services: export OPENAI_API_KEY=your-openai-key export DEEPSEEK_API_KEY=your-deepseek-key How a step picks its model The SwarmFlow guide lists model among the keys of the options bag that agent() accepts, and recommends leaving it out by default so each worker inherits the teammate model. This tutorial's pipeline does exactly that: it passes no options, so all three Teammates run on the same default model. That's the simplest setup, and it works even with a single LLM. When to split the work across models The three steps have quite different workloads: - market_researcher starts from a one-line task and has to reason across trends, competitors, and opportunities. That's open-ended work, and a larger model earns its cost here. - data_summarizer gets JSON that's already structured and returns five bullets. A smaller, faster model handles that well. - report_writer writes three long sections from both earlier outputs, so it belongs on the larger model again. So the natural refinement is to register a cheaper model under an alias in your configuration, then pass that alias on the summarizer call only: summary = await agent( "Distill this market research into exactly 5 concise, actionable bullet points:\n" + json.dumps(research), label="data_summarizer", schema=SUMMARY_SCHEMA, options={"model": "<alias>"}, ) The other two calls stay untouched and keep using the default. You cut the cost of the lightest step, and the pipeline logic doesn't change. Because the schema still checks the output, the smaller model has to produce the same five-bullet shape, or the step doesn't pass. How to run the Swarm and read the monitoring output 1. Open http://localhost:5173. 2. Choose Cluster mode in the chat box. 3. Load src/swarmflow_pipeline.py with the + button in the chat box. 4. Ask the Leader to start the Swarm, for example by entering Start the swarm. The task-progress panel then shows each phase and Teammate as it runs. !Swarm start You can also open the group chat, which shows the agents' messages to each other. !WorkSwarm: Agents' group chat A healthy run moves through Research, Summarize, and Report in order, with each [Leader] log line appearing as its phase starts. !WorkSwarm in action When ReliabilityRail escalates a problem, the Leader's response appears in the group chat, and the run's status changes in the SwarmFlow tree view. From there you can pause, resume, or stop the run. Paused runs don't resume on their own. After you send a message, the Leader decides whether to resume them. If you prefer the terminal, WorkSwarm also has a terminal UI (TUI) for macOS, Windows, and Linux, where /swarmflows opens the same run, phase, and node view. When all three Teammates finish and the workflow completes, the Leader reports back with a summary. !WorkSwarm completed with summary Where to go from here You now have a working three-step Swarm, and how it's built matters as much as what it does: - Step order lives in plain Python. One await follows another, so you can read the sequence straight off the code without depending on META. - Every handoff is schema-checked. The JSON schema on each agent() call means each step receives a validated object instead of free text it has to parse. - The whole flow runs offline. The MockBackend tests exercise all three steps, the event stream, and journal replay without an API key, so structural mistakes show up before a live run spends any tokens. - Model choice is per step. All three Teammates can share the default model, and a single options={"model": "<alias>"} moves one step to a cheaper one. From here, the WorkSwarm docs describe three directions to extend the patterns: - Human-in-the-Loop adds approval gates between phases, for workflows where a person has to sign off before the next Teammate starts. - Distributed deployment spreads the Leader and Teammates across processes and machines once a single host becomes the bottleneck. - The Experience Closed-Loop mechanism lets WorkSwarm reuse task-breakdown templates from earlier successful runs, so decomposing similar tasks gets faster over time. To get started, clone the repo, read the Agent Team and SwarmFlow guides, and launch your first Cluster-mode Swarm from openJiuwen-ai/jiuwenswarm. --- Frequently asked questions about WorkSwarm multi-agent pipelines What is WorkSwarm's Cluster mode, and how does it differ from Agent mode? In Agent mode, a single agent handles the task on its own. Cluster mode turns on multi-agent collaboration: a Leader breaks the task into subtasks and coordinates specialist Teammates, with phase tracking and ReliabilityRail failure detection across the run. Choose Cluster mode when a task splits naturally into stages that need different expertise, or different models. Is WorkSwarm the same project as JiuwenSwarm? Yes. JiuwenSwarm was renamed WorkSwarm, and the current release installs from PyPI as workswarm. The Python import package, the command-line tools, the ~/.jiuwenswarm/ config folder, and the GitHub repository still use the old name as of September 2026. How does a SwarmFlow script get into a Team session? There are three ways: - The Leader writes a script on the fly by calling its swarmflow() tool. - You install a Swarm Skill that bundles a workflow script. - You prepare a script offline and have the Leader run it with swarmflow(script_path=...). How do I reuse a SwarmFlow script across projects? Package it as a Swarm Skill with the built-in swarmskill-creator skill. You can install the resulting skill locally or publish it to the Agentic Hub so other teams can run the same workflow. Can agent calls run in parallel? Yes. SwarmFlow has parallel(), pipeline(), and map_parallel() (alias pmap()) operators for fan-out work. This tutorial uses sequential awaits because each step needs the previous step's output. How do I add a human approval step between phases? Use the human() operator for a single approval, confirmation, or choice, or human_session() for a multi-turn exchange. The run waits at that node, and you reply from the SwarmFlow tree view in the Web UI. How do I cap token spend for a Swarm? Set swarmflow_budget under modes.team.jiuwen_team in ~/.jiuwenswarm/config/config.yaml and restart the backend, or use /swarmflow on --budget in the TUI. The Web UI can switch SwarmFlow on, but it doesn't expose the budget setting. When a run exhausts its budget, it ends as failed and can't be resumed. How do I pause or stop a run that's already going? Open the SwarmFlow tree view in the Web UI and pause, resume, or stop the individual run. Paused runs don't resume on their own. After you send a message, the Leader decides whether to resume them. ---