Build a custom multi-step AI workflow graph with openJiuwen Agent Core
A multi-step LLM pipeline usually breaks at a handoff. The triage step returns "high", but the routing step expects the integer 4. Nothing raises an error. The bad value keeps moving downstream until a customer gets the wrong reply or a dashboard shows misleading figures. By then you can't say where the data went wrong, so you can't fix it quickly.
This tutorial builds a support-ticket workflow on openJiuwen Agent Core, an open-source Python SDK for agent workflows. The graph has five parts:
- A Start node.
- An LLMComponent that triages the ticket into a typed record (named fields with fixed types).
- A human approval step.
- A second LLMComponent that drafts a reply.
- An End node.
A WorkflowAgent runs the graph.
You'll also see where Agent Core enforces data contracts (rules for what each step may accept and return). The evidence comes from a proof suite that ran every scenario against the real SDK and a real model, three times in a row. Agent Core checks two boundaries at run time and leaves one unchecked. The proof suite shows that gap too.
Environment setup
You will need:
- Python 3.11, 3.12, or 3.13 (openjiuwen on PyPI declares >=3.11,<3.14)
- An OpenAI API key
- openjiuwen==0.1.19, which matches the v0.1.19 release of Agent Core
git clone https://github.com/manishh/openjiuwen--custom-multi-step-ai-workflow-agent-core-proof-first
cd openjiuwen--custom-multi-step-ai-workflow-agent-core-proof-first
cp .env.example .env # then set OPENAI_API_KEY=sk-... (no quotes, no spaces)
./prove --repeat 3
You can also export the key in your shell (export OPENAI_API_KEY=sk-...) instead of using .env. The .env file is git-ignored.
Run ./prove to start the proof (demo). It creates a virtual environment (.venv), installs the pinned dependencies from requirements.txt and constraints.txt, and runs every scenario. Each repeat runs in a new Python process and a new working directory, because Agent Core keeps its checkpointer (the store for saved workflow state) and workflow registry in process memory. A repeat makes about seven small model calls and takes about 25 seconds. ./prove prints PASS or FAIL for every check. It exits with 0 only if every check passed in every repeat.
To install the SDK into your own project, run pip install openjiuwen==0.1.19. The repo README covers the full prerequisites.
The imports this tutorial needs come from three modules. Two of them are shown below:
from openjiuwen.core.foundation.llm import ModelClientConfig, ModelRequestConfig
from openjiuwen.core.workflow import (
End,
LLMCompConfig,
LLMComponent,
Start,
Workflow,
WorkflowCard,
WorkflowComponent,
)
WorkflowAgent comes from openjiuwen.core.application.workflow_agent.
Two side effects appear on the first run. The SDK writes INFO logs to stdout, and it creates a logs/ folder in your working directory. The proof script gives each repeat its own directory and sends stdout to product.log.
WorkflowCard and workflow initialization
A WorkflowCard holds the workflow's identity and its input contract. Its input_params field takes a JSON schema. Agent Core checks every incoming input dict against that schema before any node runs.
The tutorial schema requires a query string and allows an optional integer priority_hint. The priority_hint field exists only so the proof has a typed field to break (for simulating an error).
TICKET_INPUT_SCHEMA = {
"type": "object",
"properties": {
"query": {"type": "string", "description": "The customer's ticket text"},
"priority_hint": {"type": "integer", "description": "Optional 1-5 hint from the help desk"},
},
"required": ["query"],
}
...
def _card(workflow_id: str, description: str) -> WorkflowCard:
return WorkflowCard(
id=workflow_id,
name=workflow_id,
version="1.0",
description=description,
input_params=TICKET_INPUT_SCHEMA,
)
Pass the card with the card= keyword:
flow = Workflow(card=_card("support_ticket", "Triage a ticket, get approval, draft a reply"))
> Note: The official docs show Workflow(workflow_config=WorkflowConfig(card=...)). In 0.1.19, that form runs without an error but drops the card. The workflow then gets a random ID and no input schema. Use card= until the docs catch up.
Give each graph variant its own id as well. WorkflowAgent.add_workflows registers workflows in a process-wide registry, keyed by ID and version. If you register a different graph under an existing key, the SDK logs an ERROR and the agent keeps running the first graph.
The input contract holds up in the proof. When priority_hint is "urgent", the agent reports error 100016 and no node starts. That means no model call happens:
"error_code": 100016,
"message": "workflow execute input is invalid, ... reason='input validation failed against schema: [189001] validate data with schema failed, error='1 validation error for DynamicModel\npriority_hint\n Input should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value='urgent', input_type=str]...
The schema names the required field query to match the key WorkflowAgent uses when it passes the user's message to the workflow.
Start and End node configuration
In the 0.1.19 SDK, typed inputs live on WorkflowCard.input_params. Start() accepts no arguments. Passing it a config dict raises TypeError. To connect the Start node to the workflow input, use inputs_schema.
flow.set_start_comp("start", Start(), inputs_schema={"query": "${query}"})
The ${...} strings are references. ${query} reads the workflow input. ${start.query} reads the query output of the node with ID start.
The End node has two return modes, and each one produces a different result shape. If you give End a responseTemplate, it runs in text mode. It renders a string and returns {"response": "..."}.
flow.set_end_comp(
"end",
End({"responseTemplate": "[{{category}}/{{urgency}}] escalated={{approved}} :: {{reply}}"}), # text mode
inputs_schema={"category": "${triage.category}", "urgency": "${triage.urgency}",
"approved": "${approve.approved}", "reply": "${reply.text}"},
)
If you leave out the template, End() runs in variable mode. It returns the values as a dict under output. In the proof, the variable-mode graph returned {"output": {"category": "bug", "urgency": 5}}.
Use text mode when a person reads the result. Use variable mode when code consumes it. A downstream service can read output["urgency"] directly and never has to parse a sentence.
LLM component node setup
An LLMComponent takes one LLMCompConfig. The config bundles three things: the model connection, the prompt, and the output contract.
ModelClientConfig holds the connection to the provider. ModelRequestConfig picks the model.
def model_client_config() -> ModelClientConfig:
"""How to reach the model provider. The key is read from the environment."""
return ModelClientConfig(
client_provider="OpenAI", # Agent Core's OpenAI-compatible client
api_key=os.environ["OPENAI_API_KEY"], # never hard-code or write to disk
api_base=os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1"),
verify_ssl=True, # keep TLS verification on
timeout=60, # seconds per model call
)
def model_request_config() -> ModelRequestConfig:
"""Which model to call (and, optionally, sampling settings)."""
return ModelRequestConfig(model=os.environ.get("MODEL_NAME", DEFAULT_MODEL))
The proof used gpt-5.4-mini (DEFAULT_MODEL) through the OpenAI provider, with TLS verification on. To use a different model or endpoint, set MODEL_NAME or OPENAI_BASE_URL.
The triage node combines the remaining fields. It defines the output contract and the prompts, then passes everything to LLMCompConfig.
TRIAGE_OUTPUT_CONFIG = {
"category": {"type": "string", "description": "billing | bug | other", "required": True},
"urgency": {"type": "integer", "description": "1 (low) to 5 (critical)", "required": True},
}
TRIAGE_SYSTEM_PROMPT = "You triage customer support tickets."
TRIAGE_USER_PROMPT = "Ticket: {{query}}" # {{query}} is filled from the node's inputs
...
def triage_llm(user_prompt: str = TRIAGE_USER_PROMPT, response_type: str = "json",
system_prompt: str = TRIAGE_SYSTEM_PROMPT) -> LLMComponent:
"""The triage node: classify a ticket into a typed {category, urgency} record."""
if response_type == "json":
output_config = TRIAGE_OUTPUT_CONFIG
else:
# text/markdown mode allows exactly one output field: the raw reply string.
output_config = {"triage_text": {"type": "string", "description": "raw reply", "required": True}}
return LLMComponent(LLMCompConfig(
model_client_config=model_client_config(),
model_config=model_request_config(),
template_content=[ # chat template, rendered per run
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
],
response_format={"type": response_type}, # "json" | "text" | "markdown"
output_config=output_config, # the node's output contract
))
Three fields shape what the node sends and returns:
- template_content is the chat template. Placeholders such as {{query}} are filled from the node's inputs. The graph supplies those inputs through inputs_schema={"query": "${start.query}"}.
- response_format takes {"type": "json"}, {"type": "text"} or {"type": "markdown"}.
- output_config declares the fields this node promises to downstream nodes. In JSON mode, Agent Core adds the schema to the prompt. It then parses the reply and validates it against the schema. Text and markdown modes accept exactly one output field, which holds the raw reply.
The proof tests output_config with a prompt that contradicts the schema. The schema says urgency is an integer. The prompt tells the model to report urgency as a word. This simulates someone changing the "house style" of the prompt:
"user": "Ticket: {{query}}\n\nReport urgency as one of the words low, medium or high. Never use a number for urgency."
The drift is deliberate, and gpt-5.4-mini followed the prompt in every repeat. The proof then runs the same prompt twice.
Text mode. The run completes, and "triage_text": "high" reaches End as untyped text. The exact wording varies from run to run (for example high or Urgency: high), because text mode puts no constraint on the shape of the reply. In every case, a word reaches End where the schema expects an integer, and nothing checks it.
If the model happens to answer with an integer anyway, the proof re-runs scenario C, up to three attempts. It records each retry in report.json. The checks themselves never loosen.
JSON mode. Switch triage to JSON mode with urgency: integer, and the run stops at the triage node:
"error_code": 101004,
"message": "component llm_execution process error, reason: component llm_config is invalid, reason: expected type integer but got <class 'str'>"
The node list for that run is ["start", "triage"]. End never starts, so the workflow returns no final output for a consumer to misread.
How to connect the graph and run the WorkflowAgent
The full graph adds a human approval step between triage and reply.
flowchart TD
IN([Agent input: query, priority_hint]) -->|checked against WorkflowCard.input_params| start[Start]
start -->|start.query| triage[LLMComponent triage, JSON mode, output_config]
triage -->|triage.category, triage.urgency| approve[HumanApproval, session.interact]
approve -->|approve.approved| reply[LLMComponent reply, text mode]
reply -->|reply.text| endn[End, text mode]
endn --> OUT([workflow_final])
The approval node
The approval node is a six-line custom WorkflowComponent. When it calls session.interact(), the run pauses and Agent Core saves a checkpoint (a snapshot of the workflow's state).
class HumanApproval(WorkflowComponent):
...
async def invoke(self, inputs, session, context):
answer = await session.interact(
f"Escalate this {inputs.get('category')} ticket (urgency {inputs.get('urgency')})? yes/no"
)
return {"approved": str(answer).strip().lower() == "yes"}
Registering nodes and edges
Register each node together with its input references. Then add the edges.
flow.add_workflow_comp("triage", triage_llm(), inputs_schema={"query": "${start.query}"})
flow.add_workflow_comp("approve", HumanApproval(),
inputs_schema={"category": "${triage.category}", "urgency": "${triage.urgency}"})
flow.add_workflow_comp("reply", reply_llm(),
inputs_schema={"category": "${triage.category}", "approved": "${approve.approved}"})
...
for src, dst in [("start", "triage"), ("triage", "approve"), ("approve", "reply"), ("reply", "end")]:
flow.add_connection(src, dst)
The edge check in version 0.1.19 is shallow. add_connection only confirms that the node IDs are non-empty strings. Compilation only confirms that both endpoints are registered nodes. Neither step looks at the ${...} references.
Scenario D in the proof shows the effect. It uses the smaller start → triage → end graph and changes one reference in End to ${triage.urgncy}. The run returns COMPLETED with {"output": {"category": "bug"}}. The urgency field disappears and no error appears. End drops any value that resolves to None. To cover your own references, add a unit test that asserts on the keys of the final output.
Running the WorkflowAgent
Wrap the workflow in a WorkflowAgent and stream it. The proof suite calls await Runner.start() before the agent calls and await Runner.stop() after them. The SDK's own tests use the same pattern.
agent = WorkflowAgent(WorkflowAgentConfig(
id=agent_id,
version="1.0",
description="Support-ticket agent driven by an Agent Core workflow",
workflows=[],
))
agent.add_workflows([workflow])
agent.stream() yields four kinds of chunks:
- tracer_workflow chunks, one for each node.
- tracer_agent chunks, which carry errors.
- An __interaction__ chunk, when a node pauses.
- A workflow_final chunk, which holds the result.
A failed run doesn't raise an exception from stream(), so read the error from the tracer chunks.
Turn 1: the run pauses
In turn 1, the nodes ["start", "triage", "approve"] started, and then the run paused. The output shows the triage result and the question the approval node asked:
"triage": {
"category": "billing",
"urgency": 4
}
...
"id": "approve",
"value": "Escalate this billing ticket (urgency 4)? yes/no"
In the proof, the harness answers this question itself. It resumes the run with a scripted yes (resume.update("approve", "yes") in harness/run_proof.py), so ./prove never waits for you.
Turn 2: the run resumes
To resume, send an InteractiveInput on the same conversation_id. In run_example.py, answer is whatever you type at the prompt:
resume = InteractiveInput()
resume.update(chunk.payload.id, answer) # resume that node from its checkpoint
async for chunk2 in agent.stream({"query": resume, **conv}):
Turn 2 started ["approve", "reply", "end"]. It skipped start and triage, and the final response reused turn 1's values:
"response": "[billing/4] escalated=True :: Your billing ticket has been escalated to a specialist, who will review it and follow up with you as soon as possible."
What the checkpoint covers
The checkpoint covers runs that pause for input. The checkpointer lives in memory and clears when a run completes. It handles pauses inside a live process and offers no crash recovery. If you send ordinary input to a session that still holds saved state, the SDK raises an error or clears the state.
Try it yourself
To answer the approval yourself, run the interactive example after ./prove has created .venv. The script doesn't load .env on its own, so load the key into your shell first:
set -a; source .env; set +a
.venv/bin/python run_example.py "I was charged twice this month"
The script prints each node's output and asks Escalate this billing ticket (urgency N)? yes/no. It then resumes the workflow with your answer and prints the final response. The SDK's INFO log lines appear mixed in with the output.
Error reasons and error codes
The proof suite sets one option you should think about before copying it into production. UserConfig.set_is_sensitive(False) makes the SDK show error reasons. By default, the SDK hides them and reports only "invoke llm failed". The proof suite needs the reasons to tell a model API failure (101003) from a contract failure (101004).
These are the error codes you'll see in the proof transcripts:
| Code | Meaning |
|---|---|
| 100016 | The workflow input failed the WorkflowCard.input_params schema |
| 101004 | An LLM node's JSON reply failed its output_config, for example expected type integer |
| 101003 | The model API call failed (network, auth or rate limit), which is unrelated to your contracts |
Wrapping up
You now have a WorkflowAgent that rejects malformed input with error 100016 before it spends a model call. It stops a drifted LLM reply at the node with error 101004, before the reply reaches downstream consumers. It also pauses for human approval and resumes without re-running triage. Agent Core does not check node references, so you have to cover them yourself. Add a test that asserts on the keys of the final output.
From here, try registering several workflows on one WorkflowAgent or branching the graph so independent nodes run in parallel. Clone Agent Core on GitHub to explore the SDK, and run this tutorial's example from its companion repo. If your team prefers a drag-and-drop canvas, see Agent Studio on openjiuwen.com.
---
Frequently asked questions
What does openJiuwen Agent Core do in a multi-step AI workflow?
openJiuwen Agent Core is an open-source Python SDK that lets you define agent workflows as graphs of nodes. You declare the workflow's input schema on a WorkflowCard and each LLM node's output schema in output_config. As of v0.1.19, the SDK checks the workflow input and the output of JSON-mode LLM nodes at run time. It doesn't check the references between nodes, so cover those with a test.
Why does Agent Core reject input before any node runs?
Agent Core validates every incoming input dict against the JSON schema you provide in WorkflowCard.input_params before the first node executes. If the input fails validation, the agent returns error 100016 and no model call is made, which means you don't spend tokens on a request that was already broken.
What is the difference between text mode and variable mode in the End node?
When you pass a responseTemplate to End, it renders a formatted string and returns {"response": "..."}. Without a template, End returns the raw values as a dict under output. Text mode suits human-readable results and variable mode suits downstream code that reads specific keys directly.
How does Agent Core catch a drifted LLM output?
When a node runs in JSON mode with an output_config schema, Agent Core adds that schema to the prompt, parses the model's reply, and validates the parsed result against the declared types. If the model returns a string where an integer is required, the run stops at that node with error 101004 and the downstream nodes never execute.
What happens when a node reference contains a typo?
In version 0.1.19, Agent Core's edge check is shallow. A typo such as ${triage.urgncy} resolves to None at run time, and End silently drops any value that resolves to None. The run still returns COMPLETED but the missing field is absent from the output. A unit test that asserts on the final output keys is the reliable way to catch this.
How does the human approval step pause and resume a workflow?
A custom WorkflowComponent calls await session.interact(...), which pauses the run and saves a checkpoint in memory. To resume, you send an InteractiveInput carrying the same conversation_id back to agent.stream(). The resumed run picks up from the approval node and skips the nodes that have already completed.
Why doesn't ./prove ask me to approve the escalation?
The proof answers the approval for you. To prove the same behavior on every repeat, harness/run_proof.py resumes the paused run with a scripted yes. The run still pauses at the approve node, and reply and end don't start until the answer arrives. To type the answer yourself, run run_example.py after loading your key into the shell.
Does Agent Core's checkpoint survive a process restart?
The checkpointer lives in memory and clears when a run completes or the process exits. It covers pauses inside a live process but offers no crash recovery across restarts.
Which Python versions does openJiuwen Agent Core support?
openjiuwen on PyPI declares >=3.11,<3.14, so Agent Core supports Python 3.11, 3.12, and 3.13.
Why should UserConfig.set_is_sensitive(False) stay out of production?
By default, Agent Core hides detailed error reasons and returns a generic message such as "invoke llm failed". Setting is_sensitive to False exposes the full reason string, which is useful in a proof suite but may leak internal implementation details in a production environment.
---
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…
20 views
华为系的 openJiuwen 开源了企业级 AgentOS
多智能体协同、自演进、配合一体机做小时级私有化部署。
这算是一个很明显的风向信号： 消费级 Agent 还在卷聊天和写代码，大厂体系已经在偷偷摸索国内政企最在意的“私有化交付”底座了。
开源地址：
openJiuwen-ai/agentos-enterprise
1 like · 240 views
🔗 原文: qbitai.com
💻 GitHub: openJiuwen-ai/agentos-enterprise
✍️ by ツキタマ（中国語圏AI一次情報を速さ×検証済みで届ける視点）
9 views
Mentions
Build a custom multi-step AI workflow graph with openJiuwen Agent Core A multi-step LLM pipeline usually breaks at a handoff. The triage step returns "high", but the routing step expects the integer 4. Nothing raises an error. The bad value keeps moving downstream until a customer gets the wrong reply or a dashboard shows misleading figures. By then you can't say where the data went wrong, so you can't fix it quickly. This tutorial builds a support-ticket workflow on openJiuwen Agent Core, an open-source Python SDK for agent workflows. The graph has five parts: - A Start node. - An LLMComponent that triages the ticket into a typed record (named fields with fixed types). - A human approval step. - A second LLMComponent that drafts a reply. - An End node. A WorkflowAgent runs the graph. You'll also see where Agent Core enforces data contracts (rules for what each step may accept and return). The evidence comes from a proof suite that ran every scenario against the real SDK and a real model, three times in a row. Agent Core checks two boundaries at run time and leaves one unchecked. The proof suite shows that gap too. Environment setup You will need: - Python 3.11, 3.12, or 3.13 (openjiuwen on PyPI declares >=3.11,<3.14) - An OpenAI API key - openjiuwen==0.1.19, which matches the v0.1.19 release of Agent Core git clone https://github.com/manishh/openjiuwen--custom-multi-step-ai-workflow-agent-core-proof-first cd openjiuwen--custom-multi-step-ai-workflow-agent-core-proof-first cp .env.example .env # then set OPENAI_API_KEY=sk-... (no quotes, no spaces) ./prove --repeat 3 You can also export the key in your shell (export OPENAI_API_KEY=sk-...) instead of using .env. The .env file is git-ignored. Run ./prove to start the proof (demo). It creates a virtual environment (.venv), installs the pinned dependencies from requirements.txt and constraints.txt, and runs every scenario. Each repeat runs in a new Python process and a new working directory, because Agent Core keeps its checkpointer (the store for saved workflow state) and workflow registry in process memory. A repeat makes about seven small model calls and takes about 25 seconds. ./prove prints PASS or FAIL for every check. It exits with 0 only if every check passed in every repeat. To install the SDK into your own project, run pip install openjiuwen==0.1.19. The repo README covers the full prerequisites. The imports this tutorial needs come from three modules. Two of them are shown below: from openjiuwen.core.foundation.llm import ModelClientConfig, ModelRequestConfig from openjiuwen.core.workflow import ( End, LLMCompConfig, LLMComponent, Start, Workflow, WorkflowCard, WorkflowComponent, ) WorkflowAgent comes from openjiuwen.core.application.workflow_agent. Two side effects appear on the first run. The SDK writes INFO logs to stdout, and it creates a logs/ folder in your working directory. The proof script gives each repeat its own directory and sends stdout to product.log. WorkflowCard and workflow initialization A WorkflowCard holds the workflow's identity and its input contract. Its input_params field takes a JSON schema. Agent Core checks every incoming input dict against that schema before any node runs. The tutorial schema requires a query string and allows an optional integer priority_hint. The priority_hint field exists only so the proof has a typed field to break (for simulating an error). TICKET_INPUT_SCHEMA = { "type": "object", "properties": { "query": {"type": "string", "description": "The customer's ticket text"}, "priority_hint": {"type": "integer", "description": "Optional 1-5 hint from the help desk"}, }, "required": ["query"], } ... def _card(workflow_id: str, description: str) -> WorkflowCard: return WorkflowCard( id=workflow_id, name=workflow_id, version="1.0", description=description, input_params=TICKET_INPUT_SCHEMA, ) Pass the card with the card= keyword: flow = Workflow(card=_card("support_ticket", "Triage a ticket, get approval, draft a reply")) > Note: The official docs show Workflow(workflow_config=WorkflowConfig(card=...)). In 0.1.19, that form runs without an error but drops the card. The workflow then gets a random ID and no input schema. Use card= until the docs catch up. Give each graph variant its own id as well. WorkflowAgent.add_workflows registers workflows in a process-wide registry, keyed by ID and version. If you register a different graph under an existing key, the SDK logs an ERROR and the agent keeps running the first graph. The input contract holds up in the proof. When priority_hint is "urgent", the agent reports error 100016 and no node starts. That means no model call happens: "error_code": 100016, "message": "workflow execute input is invalid, ... reason='input validation failed against schema: [189001] validate data with schema failed, error='1 validation error for DynamicModel\npriority_hint\n Input should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value='urgent', input_type=str]... The schema names the required field query to match the key WorkflowAgent uses when it passes the user's message to the workflow. Start and End node configuration In the 0.1.19 SDK, typed inputs live on WorkflowCard.input_params. Start() accepts no arguments. Passing it a config dict raises TypeError. To connect the Start node to the workflow input, use inputs_schema. flow.set_start_comp("start", Start(), inputs_schema={"query": "${query}"}) The ${...} strings are references. ${query} reads the workflow input. ${start.query} reads the query output of the node with ID start. The End node has two return modes, and each one produces a different result shape. If you give End a responseTemplate, it runs in text mode. It renders a string and returns {"response": "..."}. flow.set_end_comp( "end", End({"responseTemplate": "[{{category}}/{{urgency}}] escalated={{approved}} :: {{reply}}"}), # text mode inputs_schema={"category": "${triage.category}", "urgency": "${triage.urgency}", "approved": "${approve.approved}", "reply": "${reply.text}"}, ) If you leave out the template, End() runs in variable mode. It returns the values as a dict under output. In the proof, the variable-mode graph returned {"output": {"category": "bug", "urgency": 5}}. Use text mode when a person reads the result. Use variable mode when code consumes it. A downstream service can read output["urgency"] directly and never has to parse a sentence. LLM component node setup An LLMComponent takes one LLMCompConfig. The config bundles three things: the model connection, the prompt, and the output contract. ModelClientConfig holds the connection to the provider. ModelRequestConfig picks the model. def model_client_config() -> ModelClientConfig: """How to reach the model provider. The key is read from the environment.""" return ModelClientConfig( client_provider="OpenAI", # Agent Core's OpenAI-compatible client api_key=os.environ["OPENAI_API_KEY"], # never hard-code or write to disk api_base=os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1"), verify_ssl=True, # keep TLS verification on timeout=60, # seconds per model call ) def model_request_config() -> ModelRequestConfig: """Which model to call (and, optionally, sampling settings).""" return ModelRequestConfig(model=os.environ.get("MODEL_NAME", DEFAULT_MODEL)) The proof used gpt-5.4-mini (DEFAULT_MODEL) through the OpenAI provider, with TLS verification on. To use a different model or endpoint, set MODEL_NAME or OPENAI_BASE_URL. The triage node combines the remaining fields. It defines the output contract and the prompts, then passes everything to LLMCompConfig. TRIAGE_OUTPUT_CONFIG = { "category": {"type": "string", "description": "billing | bug | other", "required": True}, "urgency": {"type": "integer", "description": "1 (low) to 5 (critical)", "required": True}, } TRIAGE_SYSTEM_PROMPT = "You triage customer support tickets." TRIAGE_USER_PROMPT = "Ticket: {{query}}" # {{query}} is filled from the node's inputs ... def triage_llm(user_prompt: str = TRIAGE_USER_PROMPT, response_type: str = "json", system_prompt: str = TRIAGE_SYSTEM_PROMPT) -> LLMComponent: """The triage node: classify a ticket into a typed {category, urgency} record.""" if response_type == "json": output_config = TRIAGE_OUTPUT_CONFIG else: # text/markdown mode allows exactly one output field: the raw reply string. output_config = {"triage_text": {"type": "string", "description": "raw reply", "required": True}} return LLMComponent(LLMCompConfig( model_client_config=model_client_config(), model_config=model_request_config(), template_content=[ # chat template, rendered per run {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], response_format={"type": response_type}, # "json" | "text" | "markdown" output_config=output_config, # the node's output contract )) Three fields shape what the node sends and returns: - template_content is the chat template. Placeholders such as {{query}} are filled from the node's inputs. The graph supplies those inputs through inputs_schema={"query": "${start.query}"}. - response_format takes {"type": "json"}, {"type": "text"} or {"type": "markdown"}. - output_config declares the fields this node promises to downstream nodes. In JSON mode, Agent Core adds the schema to the prompt. It then parses the reply and validates it against the schema. Text and markdown modes accept exactly one output field, which holds the raw reply. The proof tests output_config with a prompt that contradicts the schema. The schema says urgency is an integer. The prompt tells the model to report urgency as a word. This simulates someone changing the "house style" of the prompt: "user": "Ticket: {{query}}\n\nReport urgency as one of the words low, medium or high. Never use a number for urgency." The drift is deliberate, and gpt-5.4-mini followed the prompt in every repeat. The proof then runs the same prompt twice. Text mode. The run completes, and "triage_text": "high" reaches End as untyped text. The exact wording varies from run to run (for example high or Urgency: high), because text mode puts no constraint on the shape of the reply. In every case, a word reaches End where the schema expects an integer, and nothing checks it. If the model happens to answer with an integer anyway, the proof re-runs scenario C, up to three attempts. It records each retry in report.json. The checks themselves never loosen. JSON mode. Switch triage to JSON mode with urgency: integer, and the run stops at the triage node: "error_code": 101004, "message": "component llm_execution process error, reason: component llm_config is invalid, reason: expected type integer but got <class 'str'>" The node list for that run is ["start", "triage"]. End never starts, so the workflow returns no final output for a consumer to misread. How to connect the graph and run the WorkflowAgent The full graph adds a human approval step between triage and reply. flowchart TD IN([Agent input: query, priority_hint]) -->|checked against WorkflowCard.input_params| start[Start] start -->|start.query| triage[LLMComponent triage, JSON mode, output_config] triage -->|triage.category, triage.urgency| approve[HumanApproval, session.interact] approve -->|approve.approved| reply[LLMComponent reply, text mode] reply -->|reply.text| endn[End, text mode] endn --> OUT([workflow_final]) The approval node The approval node is a six-line custom WorkflowComponent. When it calls session.interact(), the run pauses and Agent Core saves a checkpoint (a snapshot of the workflow's state). class HumanApproval(WorkflowComponent): ... async def invoke(self, inputs, session, context): answer = await session.interact( f"Escalate this {inputs.get('category')} ticket (urgency {inputs.get('urgency')})? yes/no" ) return {"approved": str(answer).strip().lower() == "yes"} Registering nodes and edges Register each node together with its input references. Then add the edges. flow.add_workflow_comp("triage", triage_llm(), inputs_schema={"query": "${start.query}"}) flow.add_workflow_comp("approve", HumanApproval(), inputs_schema={"category": "${triage.category}", "urgency": "${triage.urgency}"}) flow.add_workflow_comp("reply", reply_llm(), inputs_schema={"category": "${triage.category}", "approved": "${approve.approved}"}) ... for src, dst in [("start", "triage"), ("triage", "approve"), ("approve", "reply"), ("reply", "end")]: flow.add_connection(src, dst) The edge check in version 0.1.19 is shallow. add_connection only confirms that the node IDs are non-empty strings. Compilation only confirms that both endpoints are registered nodes. Neither step looks at the ${...} references. Scenario D in the proof shows the effect. It uses the smaller start → triage → end graph and changes one reference in End to ${triage.urgncy}. The run returns COMPLETED with {"output": {"category": "bug"}}. The urgency field disappears and no error appears. End drops any value that resolves to None. To cover your own references, add a unit test that asserts on the keys of the final output. Running the WorkflowAgent Wrap the workflow in a WorkflowAgent and stream it. The proof suite calls await Runner.start() before the agent calls and await Runner.stop() after them. The SDK's own tests use the same pattern. agent = WorkflowAgent(WorkflowAgentConfig( id=agent_id, version="1.0", description="Support-ticket agent driven by an Agent Core workflow", workflows=[], )) agent.add_workflows([workflow]) agent.stream() yields four kinds of chunks: - tracer_workflow chunks, one for each node. - tracer_agent chunks, which carry errors. - An __interaction__ chunk, when a node pauses. - A workflow_final chunk, which holds the result. A failed run doesn't raise an exception from stream(), so read the error from the tracer chunks. Turn 1: the run pauses In turn 1, the nodes ["start", "triage", "approve"] started, and then the run paused. The output shows the triage result and the question the approval node asked: "triage": { "category": "billing", "urgency": 4 } ... "id": "approve", "value": "Escalate this billing ticket (urgency 4)? yes/no" In the proof, the harness answers this question itself. It resumes the run with a scripted yes (resume.update("approve", "yes") in harness/run_proof.py), so ./prove never waits for you. Turn 2: the run resumes To resume, send an InteractiveInput on the same conversation_id. In run_example.py, answer is whatever you type at the prompt: resume = InteractiveInput() resume.update(chunk.payload.id, answer) # resume that node from its checkpoint async for chunk2 in agent.stream({"query": resume, **conv}): Turn 2 started ["approve", "reply", "end"]. It skipped start and triage, and the final response reused turn 1's values: "response": "[billing/4] escalated=True :: Your billing ticket has been escalated to a specialist, who will review it and follow up with you as soon as possible." What the checkpoint covers The checkpoint covers runs that pause for input. The checkpointer lives in memory and clears when a run completes. It handles pauses inside a live process and offers no crash recovery. If you send ordinary input to a session that still holds saved state, the SDK raises an error or clears the state. Try it yourself To answer the approval yourself, run the interactive example after ./prove has created .venv. The script doesn't load .env on its own, so load the key into your shell first: set -a; source .env; set +a .venv/bin/python run_example.py "I was charged twice this month" The script prints each node's output and asks Escalate this billing ticket (urgency N)? yes/no. It then resumes the workflow with your answer and prints the final response. The SDK's INFO log lines appear mixed in with the output. Error reasons and error codes The proof suite sets one option you should think about before copying it into production. UserConfig.set_is_sensitive(False) makes the SDK show error reasons. By default, the SDK hides them and reports only "invoke llm failed". The proof suite needs the reasons to tell a model API failure (101003) from a contract failure (101004). These are the error codes you'll see in the proof transcripts: | Code | Meaning | |---|---| | 100016 | The workflow input failed the WorkflowCard.input_params schema | | 101004 | An LLM node's JSON reply failed its output_config, for example expected type integer | | 101003 | The model API call failed (network, auth or rate limit), which is unrelated to your contracts | Wrapping up You now have a WorkflowAgent that rejects malformed input with error 100016 before it spends a model call. It stops a drifted LLM reply at the node with error 101004, before the reply reaches downstream consumers. It also pauses for human approval and resumes without re-running triage. Agent Core does not check node references, so you have to cover them yourself. Add a test that asserts on the keys of the final output. From here, try registering several workflows on one WorkflowAgent or branching the graph so independent nodes run in parallel. Clone Agent Core on GitHub to explore the SDK, and run this tutorial's example from its companion repo. If your team prefers a drag-and-drop canvas, see Agent Studio on openjiuwen.com. --- Frequently asked questions What does openJiuwen Agent Core do in a multi-step AI workflow? openJiuwen Agent Core is an open-source Python SDK that lets you define agent workflows as graphs of nodes. You declare the workflow's input schema on a WorkflowCard and each LLM node's output schema in output_config. As of v0.1.19, the SDK checks the workflow input and the output of JSON-mode LLM nodes at run time. It doesn't check the references between nodes, so cover those with a test. Why does Agent Core reject input before any node runs? Agent Core validates every incoming input dict against the JSON schema you provide in WorkflowCard.input_params before the first node executes. If the input fails validation, the agent returns error 100016 and no model call is made, which means you don't spend tokens on a request that was already broken. What is the difference between text mode and variable mode in the End node? When you pass a responseTemplate to End, it renders a formatted string and returns {"response": "..."}. Without a template, End returns the raw values as a dict under output. Text mode suits human-readable results and variable mode suits downstream code that reads specific keys directly. How does Agent Core catch a drifted LLM output? When a node runs in JSON mode with an output_config schema, Agent Core adds that schema to the prompt, parses the model's reply, and validates the parsed result against the declared types. If the model returns a string where an integer is required, the run stops at that node with error 101004 and the downstream nodes never execute. What happens when a node reference contains a typo? In version 0.1.19, Agent Core's edge check is shallow. A typo such as ${triage.urgncy} resolves to None at run time, and End silently drops any value that resolves to None. The run still returns COMPLETED but the missing field is absent from the output. A unit test that asserts on the final output keys is the reliable way to catch this. How does the human approval step pause and resume a workflow? A custom WorkflowComponent calls await session.interact(...), which pauses the run and saves a checkpoint in memory. To resume, you send an InteractiveInput carrying the same conversation_id back to agent.stream(). The resumed run picks up from the approval node and skips the nodes that have already completed. Why doesn't ./prove ask me to approve the escalation? The proof answers the approval for you. To prove the same behavior on every repeat, harness/run_proof.py resumes the paused run with a scripted yes. The run still pauses at the approve node, and reply and end don't start until the answer arrives. To type the answer yourself, run run_example.py after loading your key into the shell. Does Agent Core's checkpoint survive a process restart? The checkpointer lives in memory and clears when a run completes or the process exits. It covers pauses inside a live process but offers no crash recovery across restarts. Which Python versions does openJiuwen Agent Core support? openjiuwen on PyPI declares >=3.11,<3.14, so Agent Core supports Python 3.11, 3.12, and 3.13. Why should UserConfig.set_is_sensitive(False) stay out of production? By default, Agent Core hides detailed error reasons and returns a generic message such as "invoke llm failed". Setting is_sensitive to False exposes the full reason string, which is useful in a proof suite but may leak internal implementation details in a production environment. ---
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…
华为系的 openJiuwen 开源了企业级 AgentOS 多智能体协同、自演进、配合一体机做小时级私有化部署。 这算是一个很明显的风向信号： 消费级 Agent 还在卷聊天和写代码，大厂体系已经在偷偷摸索国内政企最在意的“私有化交付”底座了。 开源地址： openJiuwen-ai/agentos-enterprise
🔗 原文: qbitai.com 💻 GitHub: openJiuwen-ai/agentos-enterprise ✍️ by ツキタマ（中国語圏AI一次情報を速さ×検証済みで届ける視点）