DEV Community

Manish Hatwalne
Manish Hatwalne

Posted on

Build a custom multi-step AI workflow graph with openJiuwen Agent Core

#ai #agents #openjiuwen

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:

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
Enter fullscreen mode Exit fullscreen mode

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,
)
Enter fullscreen mode Exit fullscreen mode

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,
    )
Enter fullscreen mode Exit fullscreen mode

Pass the card with the card= keyword:

flow = Workflow(card=_card("support_ticket", "Triage a ticket, get approval, draft a reply"))
Enter fullscreen mode Exit fullscreen mode

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]...
Enter fullscreen mode Exit fullscreen mode

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}"})
Enter fullscreen mode Exit fullscreen mode

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}"},
)
Enter fullscreen mode Exit fullscreen mode

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))
Enter fullscreen mode Exit fullscreen mode

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
    ))
Enter fullscreen mode Exit fullscreen mode

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."
Enter fullscreen mode Exit fullscreen mode

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'>"
Enter fullscreen mode Exit fullscreen mode

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"}
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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])
Enter fullscreen mode Exit fullscreen mode

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"
Enter fullscreen mode Exit fullscreen mode

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}):
Enter fullscreen mode Exit fullscreen mode

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."
Enter fullscreen mode Exit fullscreen mode

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"
Enter fullscreen mode Exit fullscreen mode

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.

Top comments (0)