ReAct Agent: Streaming and Provenance Recall

What It Is

ReAct is a vocabulary for an agent loop Dagonizer already represents as a DAG: thought, action, observation, and final answer are observable stages in the registered graph. The engine does not need a separate agent runtime to support it.

This guide maps the 8-node loop onto ReAct, then shows how the runnable examples stream reasoning traces with DagStreamProducer, stream live model deltas with chatStream, and record/recall reasoning with RDF provenance.

How It Works

The eight-node agent loop is a DAG. ReAct labels the loop's observable stages as thought, action, observation, and final. Dagonizer streams those stages through Execution, DagStreamProducer, and StreamChannel, then records and routes them with ordinary scatter DAGs.

ReAct (Reason + Act) names a loop already built into @studnicky/dagonizer: the 8-node agent loop authored as JSON-LD. This guide gives that loop its ReAct vocabulary, then adds three capabilities on top of surfaces the engine already ships: streaming (streaming producers) and graph provenance. None of this is a new execution mechanism.

Diagrams, Examples, and Outputs

The runnable ReAct examples expose three DAGs: the inner agent loop, the reasoning-trace memory writer, and the routed stream sink:

ReAct agent loop DAG

12 placements

DAG JSON-LD registered with the dispatcher

{
  "@context": {
    "@version": 1.1,
    "name": {
      "@id": "https://noocodec.dev/ontology/dag/name"
    },
    "version": {
      "@id": "https://noocodec.dev/ontology/dag/version"
    },
    "entrypoints": {
      "@id": "https://noocodec.dev/ontology/dag/entrypoints",
      "@container": "@index"
    },
    "nodes": {
      "@id": "https://noocodec.dev/ontology/dag/nodes",
      "@container": "@set"
    },
    "outputs": {
      "@id": "https://noocodec.dev/ontology/dag/outputs"
    },
    "node": {
      "@id": "https://noocodec.dev/ontology/dag/node"
    },
    "dag": {
      "@id": "https://noocodec.dev/ontology/dag/dag"
    },
    "body": {
      "@id": "https://noocodec.dev/ontology/dag/body"
    },
    "source": {
      "@id": "https://noocodec.dev/ontology/dag/source"
    },
    "sources": {
      "@id": "https://noocodec.dev/ontology/dag/sources",
      "@container": "@index"
    },
    "itemKey": {
      "@id": "https://noocodec.dev/ontology/dag/itemKey"
    },
    "execution": {
      "@id": "https://noocodec.dev/ontology/dag/execution"
    },
    "concurrency": {
      "@id": "https://noocodec.dev/ontology/dag/concurrency"
    },
    "throttle": {
      "@id": "https://noocodec.dev/ontology/dag/throttle"
    },
    "reservoir": {
      "@id": "https://noocodec.dev/ontology/dag/reservoir"
    },
    "gather": {
      "@id": "https://noocodec.dev/ontology/dag/gather"
    },
    "dagReference": {
      "@id": "https://noocodec.dev/ontology/dag/dagReference",
      "@type": "@id"
    },
    "DagReference": {
      "@id": "https://noocodec.dev/ontology/dag/DagReference"
    },
    "from": {
      "@id": "https://noocodec.dev/ontology/dag/from"
    },
    "path": {
      "@id": "https://noocodec.dev/ontology/dag/path"
    },
    "candidates": {
      "@id": "https://noocodec.dev/ontology/dag/candidates",
      "@container": "@set"
    },
    "candidateDag": {
      "@id": "https://noocodec.dev/ontology/dag/candidateDag",
      "@type": "@id"
    },
    "selectedDag": {
      "@id": "https://noocodec.dev/ontology/dag/selectedDag",
      "@type": "@id"
    },
    "resultField": {
      "@id": "https://noocodec.dev/ontology/dag/resultField"
    },
    "policy": {
      "@id": "https://noocodec.dev/ontology/dag/policy"
    },
    "reducer": {
      "@id": "https://noocodec.dev/ontology/dag/reducer"
    },
    "outcome": {
      "@id": "https://noocodec.dev/ontology/dag/outcome"
    },
    "phase": {
      "@id": "https://noocodec.dev/ontology/dag/phase"
    },
    "stateMapping": {
      "@id": "https://noocodec.dev/ontology/dag/stateMapping"
    },
    "container": {
      "@id": "https://noocodec.dev/ontology/dag/container"
    },
    "DAG": {
      "@id": "https://noocodec.dev/ontology/dag/DAG"
    },
    "Placement": {
      "@id": "https://noocodec.dev/ontology/dag/Placement"
    },
    "SingleNode": {
      "@id": "https://noocodec.dev/ontology/dag/SingleNode"
    },
    "ScatterNode": {
      "@id": "https://noocodec.dev/ontology/dag/ScatterNode"
    },
    "EmbeddedDAGNode": {
      "@id": "https://noocodec.dev/ontology/dag/EmbeddedDAGNode"
    },
    "GatherNode": {
      "@id": "https://noocodec.dev/ontology/dag/GatherNode"
    },
    "TerminalNode": {
      "@id": "https://noocodec.dev/ontology/dag/TerminalNode"
    },
    "PhaseNode": {
      "@id": "https://noocodec.dev/ontology/dag/PhaseNode"
    }
  },
  "@id": "urn:noocodec:dag:react-agent",
  "@type": "DAG",
  "name": "react-agent",
  "version": "1",
  "entrypoints": {
    "main": "urn:noocodec:dag:react-agent/node/build-request"
  },
  "nodes": [
    {
      "@id": "urn:noocodec:dag:react-agent/node/build-request",
      "@type": "SingleNode",
      "name": "build-request",
      "node": "urn:noocodec:node:build-request",
      "outputs": {
        "ready": "urn:noocodec:dag:react-agent/node/call-model",
        "error": "urn:noocodec:dag:react-agent/node/end-error"
      }
    },
    {
      "@id": "urn:noocodec:dag:react-agent/node/call-model",
      "@type": "SingleNode",
      "name": "call-model",
      "node": "urn:noocodec:node:call-model",
      "outputs": {
        "text": "urn:noocodec:dag:react-agent/node/normalize-response",
        "tools": "urn:noocodec:dag:react-agent/node/normalize-response",
        "mixed": "urn:noocodec:dag:react-agent/node/normalize-response",
        "error": "urn:noocodec:dag:react-agent/node/end-error"
      }
    },
    {
      "@id": "urn:noocodec:dag:react-agent/node/normalize-response",
      "@type": "SingleNode",
      "name": "normalize-response",
      "node": "urn:noocodec:node:normalize-response",
      "outputs": {
        "text": "urn:noocodec:dag:react-agent/node/append-assistant",
        "tools": "urn:noocodec:dag:react-agent/node/decode-tools",
        "mixed": "urn:noocodec:dag:react-agent/node/decode-tools",
        "empty": "urn:noocodec:dag:react-agent/node/end-error",
        "error": "urn:noocodec:dag:react-agent/node/end-error"
      }
    },
    {
      "@id": "urn:noocodec:dag:react-agent/node/append-assistant",
      "@type": "SingleNode",
      "name": "append-assistant",
      "node": "urn:noocodec:node:append-assistant",
      "outputs": {
        "done": "urn:noocodec:dag:react-agent/node/end-done",
        "error": "urn:noocodec:dag:react-agent/node/end-error"
      }
    },
    {
      "@id": "urn:noocodec:dag:react-agent/node/decode-tools",
      "@type": "SingleNode",
      "name": "decode-tools",
      "node": "urn:noocodec:node:decode-tools",
      "outputs": {
        "decoded": "urn:noocodec:dag:react-agent/node/normalize-tools",
        "empty": "urn:noocodec:dag:react-agent/node/end-error",
        "error": "urn:noocodec:dag:react-agent/node/end-error"
      }
    },
    {
      "@id": "urn:noocodec:dag:react-agent/node/normalize-tools",
      "@type": "SingleNode",
      "name": "normalize-tools",
      "node": "urn:noocodec:node:normalize-tools",
      "outputs": {
        "valid": "urn:noocodec:dag:react-agent/node/worksets",
        "empty": "urn:noocodec:dag:react-agent/node/end-error",
        "error": "urn:noocodec:dag:react-agent/node/end-error"
      }
    },
    {
      "@id": "urn:noocodec:dag:react-agent/node/worksets",
      "@type": "SingleNode",
      "name": "worksets",
      "node": "urn:noocodec:node:build-worksets",
      "outputs": {
        "ready": "urn:noocodec:dag:react-agent/node/dispatch-tools",
        "empty": "urn:noocodec:dag:react-agent/node/end-error",
        "error": "urn:noocodec:dag:react-agent/node/end-error"
      }
    },
    {
      "@id": "urn:noocodec:dag:react-agent/node/dispatch-tools",
      "@type": "ScatterNode",
      "name": "dispatch-tools",
      "source": "safeWorkset",
      "body": {
        "dag": {
          "@type": "DagReference",
          "from": "item",
          "path": "dagIri",
          "candidates": [
            "urn:noocodec:tool:lookup"
          ]
        }
      },
      "outputs": {
        "all-success": "urn:noocodec:dag:react-agent/node/join-tool-results",
        "partial": "urn:noocodec:dag:react-agent/node/join-tool-results",
        "all-error": "urn:noocodec:dag:react-agent/node/join-tool-results",
        "empty": "urn:noocodec:dag:react-agent/node/join-tool-results"
      },
      "itemKey": "currentItem",
      "reducer": "aggregate"
    },
    {
      "@id": "urn:noocodec:dag:react-agent/node/join-tool-results",
      "@type": "GatherNode",
      "name": "join-tool-results",
      "sources": {
        "urn:noocodec:dag:react-agent/node/dispatch-tools": {}
      },
      "gather": {
        "strategy": "map",
        "mapping": {
          "output": "toolOutputs"
        }
      },
      "outputs": {
        "success": "urn:noocodec:dag:react-agent/node/collect-results",
        "error": "urn:noocodec:dag:react-agent/node/end-error",
        "empty": "urn:noocodec:dag:react-agent/node/collect-results"
      }
    },
    {
      "@id": "urn:noocodec:dag:react-agent/node/collect-results",
      "@type": "SingleNode",
      "name": "collect-results",
      "node": "urn:noocodec:node:collect-results",
      "outputs": {
        "done": "urn:noocodec:dag:react-agent/node/build-request",
        "empty": "urn:noocodec:dag:react-agent/node/build-request",
        "error": "urn:noocodec:dag:react-agent/node/end-error"
      }
    },
    {
      "@id": "urn:noocodec:dag:react-agent/node/end-done",
      "@type": "TerminalNode",
      "name": "end-done",
      "outcome": "completed"
    },
    {
      "@id": "urn:noocodec:dag:react-agent/node/end-error",
      "@type": "TerminalNode",
      "name": "end-error",
      "outcome": "failed"
    }
  ]
}

Mermaid generated from the same DAG

Mermaid source

%%{init: {"flowchart":{"nodeSpacing":92,"rankSpacing":104,"padding":28}}}%%
flowchart TB
  %% react-agent (v1)
  entry_main(["main"])
  entry_main --> urn_noocodec_dag_react-agent/node/build-request
  urn_noocodec_dag_react-agent/node/build-request["build-request"]
  urn_noocodec_dag_react-agent/node/build-request -->|ready| urn_noocodec_dag_react-agent/node/call-model
  urn_noocodec_dag_react-agent/node/build-request -->|error| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/call-model["call-model"]
  urn_noocodec_dag_react-agent/node/call-model -->|text| urn_noocodec_dag_react-agent/node/normalize-response
  urn_noocodec_dag_react-agent/node/call-model -->|tools| urn_noocodec_dag_react-agent/node/normalize-response
  urn_noocodec_dag_react-agent/node/call-model -->|mixed| urn_noocodec_dag_react-agent/node/normalize-response
  urn_noocodec_dag_react-agent/node/call-model -->|error| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/normalize-response["normalize-response"]
  urn_noocodec_dag_react-agent/node/normalize-response -->|text| urn_noocodec_dag_react-agent/node/append-assistant
  urn_noocodec_dag_react-agent/node/normalize-response -->|tools| urn_noocodec_dag_react-agent/node/decode-tools
  urn_noocodec_dag_react-agent/node/normalize-response -->|mixed| urn_noocodec_dag_react-agent/node/decode-tools
  urn_noocodec_dag_react-agent/node/normalize-response -->|empty| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/normalize-response -->|error| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/append-assistant["append-assistant"]
  urn_noocodec_dag_react-agent/node/append-assistant -->|done| urn_noocodec_dag_react-agent/node/end-done
  urn_noocodec_dag_react-agent/node/append-assistant -->|error| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/decode-tools["decode-tools"]
  urn_noocodec_dag_react-agent/node/decode-tools -->|decoded| urn_noocodec_dag_react-agent/node/normalize-tools
  urn_noocodec_dag_react-agent/node/decode-tools -->|empty| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/decode-tools -->|error| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/normalize-tools["normalize-tools"]
  urn_noocodec_dag_react-agent/node/normalize-tools -->|valid| urn_noocodec_dag_react-agent/node/worksets
  urn_noocodec_dag_react-agent/node/normalize-tools -->|empty| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/normalize-tools -->|error| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/worksets["worksets"]
  urn_noocodec_dag_react-agent/node/worksets -->|ready| urn_noocodec_dag_react-agent/node/dispatch-tools
  urn_noocodec_dag_react-agent/node/worksets -->|empty| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/worksets -->|error| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/dispatch-tools[/"dispatch-tools"/]
  urn_noocodec_dag_react-agent/node/dispatch-tools -->|all-success| urn_noocodec_dag_react-agent/node/join-tool-results
  urn_noocodec_dag_react-agent/node/dispatch-tools -->|partial| urn_noocodec_dag_react-agent/node/join-tool-results
  urn_noocodec_dag_react-agent/node/dispatch-tools -->|all-error| urn_noocodec_dag_react-agent/node/join-tool-results
  urn_noocodec_dag_react-agent/node/dispatch-tools -->|empty| urn_noocodec_dag_react-agent/node/join-tool-results
  urn_noocodec_dag_react-agent/node/join-tool-results{"join-tool-results"}
  urn_noocodec_dag_react-agent/node/join-tool-results -->|success| urn_noocodec_dag_react-agent/node/collect-results
  urn_noocodec_dag_react-agent/node/join-tool-results -->|error| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/join-tool-results -->|empty| urn_noocodec_dag_react-agent/node/collect-results
  urn_noocodec_dag_react-agent/node/collect-results["collect-results"]
  urn_noocodec_dag_react-agent/node/collect-results -->|done| urn_noocodec_dag_react-agent/node/build-request
  urn_noocodec_dag_react-agent/node/collect-results -->|empty| urn_noocodec_dag_react-agent/node/build-request
  urn_noocodec_dag_react-agent/node/collect-results -->|error| urn_noocodec_dag_react-agent/node/end-error
  urn_noocodec_dag_react-agent/node/end-done((("end-done")))
  urn_noocodec_dag_react-agent/node/end-error>"end-error"]

ReAct reasoning trace memory DAG

3 placements

DAG JSON-LD registered with the dispatcher

{
  "@context": {
    "@version": 1.1,
    "name": {
      "@id": "https://noocodec.dev/ontology/dag/name"
    },
    "version": {
      "@id": "https://noocodec.dev/ontology/dag/version"
    },
    "entrypoints": {
      "@id": "https://noocodec.dev/ontology/dag/entrypoints",
      "@container": "@index"
    },
    "nodes": {
      "@id": "https://noocodec.dev/ontology/dag/nodes",
      "@container": "@set"
    },
    "outputs": {
      "@id": "https://noocodec.dev/ontology/dag/outputs"
    },
    "node": {
      "@id": "https://noocodec.dev/ontology/dag/node"
    },
    "dag": {
      "@id": "https://noocodec.dev/ontology/dag/dag"
    },
    "body": {
      "@id": "https://noocodec.dev/ontology/dag/body"
    },
    "source": {
      "@id": "https://noocodec.dev/ontology/dag/source"
    },
    "sources": {
      "@id": "https://noocodec.dev/ontology/dag/sources",
      "@container": "@index"
    },
    "itemKey": {
      "@id": "https://noocodec.dev/ontology/dag/itemKey"
    },
    "execution": {
      "@id": "https://noocodec.dev/ontology/dag/execution"
    },
    "concurrency": {
      "@id": "https://noocodec.dev/ontology/dag/concurrency"
    },
    "throttle": {
      "@id": "https://noocodec.dev/ontology/dag/throttle"
    },
    "reservoir": {
      "@id": "https://noocodec.dev/ontology/dag/reservoir"
    },
    "gather": {
      "@id": "https://noocodec.dev/ontology/dag/gather"
    },
    "dagReference": {
      "@id": "https://noocodec.dev/ontology/dag/dagReference",
      "@type": "@id"
    },
    "DagReference": {
      "@id": "https://noocodec.dev/ontology/dag/DagReference"
    },
    "from": {
      "@id": "https://noocodec.dev/ontology/dag/from"
    },
    "path": {
      "@id": "https://noocodec.dev/ontology/dag/path"
    },
    "candidates": {
      "@id": "https://noocodec.dev/ontology/dag/candidates",
      "@container": "@set"
    },
    "candidateDag": {
      "@id": "https://noocodec.dev/ontology/dag/candidateDag",
      "@type": "@id"
    },
    "selectedDag": {
      "@id": "https://noocodec.dev/ontology/dag/selectedDag",
      "@type": "@id"
    },
    "resultField": {
      "@id": "https://noocodec.dev/ontology/dag/resultField"
    },
    "policy": {
      "@id": "https://noocodec.dev/ontology/dag/policy"
    },
    "reducer": {
      "@id": "https://noocodec.dev/ontology/dag/reducer"
    },
    "outcome": {
      "@id": "https://noocodec.dev/ontology/dag/outcome"
    },
    "phase": {
      "@id": "https://noocodec.dev/ontology/dag/phase"
    },
    "stateMapping": {
      "@id": "https://noocodec.dev/ontology/dag/stateMapping"
    },
    "container": {
      "@id": "https://noocodec.dev/ontology/dag/container"
    },
    "DAG": {
      "@id": "https://noocodec.dev/ontology/dag/DAG"
    },
    "Placement": {
      "@id": "https://noocodec.dev/ontology/dag/Placement"
    },
    "SingleNode": {
      "@id": "https://noocodec.dev/ontology/dag/SingleNode"
    },
    "ScatterNode": {
      "@id": "https://noocodec.dev/ontology/dag/ScatterNode"
    },
    "EmbeddedDAGNode": {
      "@id": "https://noocodec.dev/ontology/dag/EmbeddedDAGNode"
    },
    "GatherNode": {
      "@id": "https://noocodec.dev/ontology/dag/GatherNode"
    },
    "TerminalNode": {
      "@id": "https://noocodec.dev/ontology/dag/TerminalNode"
    },
    "PhaseNode": {
      "@id": "https://noocodec.dev/ontology/dag/PhaseNode"
    }
  },
  "@id": "urn:noocodec:dag:react-agent-memory-trace",
  "@type": "DAG",
  "name": "react-agent-memory-trace",
  "version": "1",
  "entrypoints": {
    "main": "urn:noocodec:dag:react-agent-memory-trace/node/scatter-steps"
  },
  "nodes": [
    {
      "@id": "urn:noocodec:dag:react-agent-memory-trace/node/scatter-steps",
      "@type": "ScatterNode",
      "name": "scatter-steps",
      "body": {
        "node": "urn:noocodec:node:record-step"
      },
      "source": "source",
      "itemKey": "reasoning-step-item",
      "execution": {
        "mode": "item",
        "concurrency": 1
      },
      "outputs": {
        "all-success": "urn:noocodec:dag:react-agent-memory-trace/node/collect-steps",
        "partial": "urn:noocodec:dag:react-agent-memory-trace/node/collect-steps",
        "all-error": "urn:noocodec:dag:react-agent-memory-trace/node/collect-steps",
        "empty": "urn:noocodec:dag:react-agent-memory-trace/node/end"
      }
    },
    {
      "@id": "urn:noocodec:dag:react-agent-memory-trace/node/collect-steps",
      "@type": "GatherNode",
      "name": "collect-steps",
      "sources": {
        "urn:noocodec:dag:react-agent-memory-trace/node/scatter-steps": {}
      },
      "gather": {
        "strategy": "append",
        "target": "steps"
      },
      "outputs": {
        "success": "urn:noocodec:dag:react-agent-memory-trace/node/end",
        "error": "urn:noocodec:dag:react-agent-memory-trace/node/end",
        "empty": "urn:noocodec:dag:react-agent-memory-trace/node/end"
      }
    },
    {
      "@id": "urn:noocodec:dag:react-agent-memory-trace/node/end",
      "@type": "TerminalNode",
      "name": "end",
      "outcome": "completed"
    }
  ]
}

Mermaid generated from the same DAG

Mermaid source

%%{init: {"flowchart":{"nodeSpacing":92,"rankSpacing":104,"padding":28}}}%%
flowchart TB
  %% react-agent-memory-trace (v1)
  entry_main(["main"])
  entry_main --> urn_noocodec_dag_react-agent-memory-trace/node/scatter-steps
  urn_noocodec_dag_react-agent-memory-trace/node/scatter-steps[/"scatter-steps"/]
  urn_noocodec_dag_react-agent-memory-trace/node/scatter-steps -->|all-success| urn_noocodec_dag_react-agent-memory-trace/node/collect-steps
  urn_noocodec_dag_react-agent-memory-trace/node/scatter-steps -->|partial| urn_noocodec_dag_react-agent-memory-trace/node/collect-steps
  urn_noocodec_dag_react-agent-memory-trace/node/scatter-steps -->|all-error| urn_noocodec_dag_react-agent-memory-trace/node/collect-steps
  urn_noocodec_dag_react-agent-memory-trace/node/scatter-steps -->|empty| urn_noocodec_dag_react-agent-memory-trace/node/end
  urn_noocodec_dag_react-agent-memory-trace/node/collect-steps{"collect-steps"}
  urn_noocodec_dag_react-agent-memory-trace/node/collect-steps -->|success| urn_noocodec_dag_react-agent-memory-trace/node/end
  urn_noocodec_dag_react-agent-memory-trace/node/collect-steps -->|error| urn_noocodec_dag_react-agent-memory-trace/node/end
  urn_noocodec_dag_react-agent-memory-trace/node/collect-steps -->|empty| urn_noocodec_dag_react-agent-memory-trace/node/end
  urn_noocodec_dag_react-agent-memory-trace/node/end((("end")))

ReAct routed stream sink DAG

3 placements

DAG JSON-LD registered with the dispatcher

{
  "@context": {
    "@version": 1.1,
    "name": {
      "@id": "https://noocodec.dev/ontology/dag/name"
    },
    "version": {
      "@id": "https://noocodec.dev/ontology/dag/version"
    },
    "entrypoints": {
      "@id": "https://noocodec.dev/ontology/dag/entrypoints",
      "@container": "@index"
    },
    "nodes": {
      "@id": "https://noocodec.dev/ontology/dag/nodes",
      "@container": "@set"
    },
    "outputs": {
      "@id": "https://noocodec.dev/ontology/dag/outputs"
    },
    "node": {
      "@id": "https://noocodec.dev/ontology/dag/node"
    },
    "dag": {
      "@id": "https://noocodec.dev/ontology/dag/dag"
    },
    "body": {
      "@id": "https://noocodec.dev/ontology/dag/body"
    },
    "source": {
      "@id": "https://noocodec.dev/ontology/dag/source"
    },
    "sources": {
      "@id": "https://noocodec.dev/ontology/dag/sources",
      "@container": "@index"
    },
    "itemKey": {
      "@id": "https://noocodec.dev/ontology/dag/itemKey"
    },
    "execution": {
      "@id": "https://noocodec.dev/ontology/dag/execution"
    },
    "concurrency": {
      "@id": "https://noocodec.dev/ontology/dag/concurrency"
    },
    "throttle": {
      "@id": "https://noocodec.dev/ontology/dag/throttle"
    },
    "reservoir": {
      "@id": "https://noocodec.dev/ontology/dag/reservoir"
    },
    "gather": {
      "@id": "https://noocodec.dev/ontology/dag/gather"
    },
    "dagReference": {
      "@id": "https://noocodec.dev/ontology/dag/dagReference",
      "@type": "@id"
    },
    "DagReference": {
      "@id": "https://noocodec.dev/ontology/dag/DagReference"
    },
    "from": {
      "@id": "https://noocodec.dev/ontology/dag/from"
    },
    "path": {
      "@id": "https://noocodec.dev/ontology/dag/path"
    },
    "candidates": {
      "@id": "https://noocodec.dev/ontology/dag/candidates",
      "@container": "@set"
    },
    "candidateDag": {
      "@id": "https://noocodec.dev/ontology/dag/candidateDag",
      "@type": "@id"
    },
    "selectedDag": {
      "@id": "https://noocodec.dev/ontology/dag/selectedDag",
      "@type": "@id"
    },
    "resultField": {
      "@id": "https://noocodec.dev/ontology/dag/resultField"
    },
    "policy": {
      "@id": "https://noocodec.dev/ontology/dag/policy"
    },
    "reducer": {
      "@id": "https://noocodec.dev/ontology/dag/reducer"
    },
    "outcome": {
      "@id": "https://noocodec.dev/ontology/dag/outcome"
    },
    "phase": {
      "@id": "https://noocodec.dev/ontology/dag/phase"
    },
    "stateMapping": {
      "@id": "https://noocodec.dev/ontology/dag/stateMapping"
    },
    "container": {
      "@id": "https://noocodec.dev/ontology/dag/container"
    },
    "DAG": {
      "@id": "https://noocodec.dev/ontology/dag/DAG"
    },
    "Placement": {
      "@id": "https://noocodec.dev/ontology/dag/Placement"
    },
    "SingleNode": {
      "@id": "https://noocodec.dev/ontology/dag/SingleNode"
    },
    "ScatterNode": {
      "@id": "https://noocodec.dev/ontology/dag/ScatterNode"
    },
    "EmbeddedDAGNode": {
      "@id": "https://noocodec.dev/ontology/dag/EmbeddedDAGNode"
    },
    "GatherNode": {
      "@id": "https://noocodec.dev/ontology/dag/GatherNode"
    },
    "TerminalNode": {
      "@id": "https://noocodec.dev/ontology/dag/TerminalNode"
    },
    "PhaseNode": {
      "@id": "https://noocodec.dev/ontology/dag/PhaseNode"
    }
  },
  "@id": "urn:noocodec:dag:react-agent-routing",
  "@type": "DAG",
  "name": "react-agent-routing",
  "version": "1",
  "entrypoints": {
    "main": "urn:noocodec:dag:react-agent-routing/node/scatter-chunks"
  },
  "nodes": [
    {
      "@id": "urn:noocodec:dag:react-agent-routing/node/scatter-chunks",
      "@type": "ScatterNode",
      "name": "scatter-chunks",
      "body": {
        "node": "urn:noocodec:node:route-chunk"
      },
      "source": "source",
      "itemKey": "routed-chunk-item",
      "execution": {
        "mode": "item",
        "concurrency": 4
      },
      "outputs": {
        "all-success": "urn:noocodec:dag:react-agent-routing/node/collect-chunks",
        "partial": "urn:noocodec:dag:react-agent-routing/node/collect-chunks",
        "all-error": "urn:noocodec:dag:react-agent-routing/node/collect-chunks",
        "empty": "urn:noocodec:dag:react-agent-routing/node/end"
      }
    },
    {
      "@id": "urn:noocodec:dag:react-agent-routing/node/collect-chunks",
      "@type": "GatherNode",
      "name": "collect-chunks",
      "sources": {
        "urn:noocodec:dag:react-agent-routing/node/scatter-chunks": {}
      },
      "gather": {
        "strategy": "append",
        "target": "chunks"
      },
      "outputs": {
        "success": "urn:noocodec:dag:react-agent-routing/node/end",
        "error": "urn:noocodec:dag:react-agent-routing/node/end",
        "empty": "urn:noocodec:dag:react-agent-routing/node/end"
      }
    },
    {
      "@id": "urn:noocodec:dag:react-agent-routing/node/end",
      "@type": "TerminalNode",
      "name": "end",
      "outcome": "completed"
    }
  ]
}

Mermaid generated from the same DAG

Mermaid source

%%{init: {"flowchart":{"nodeSpacing":92,"rankSpacing":104,"padding":28}}}%%
flowchart TB
  %% react-agent-routing (v1)
  entry_main(["main"])
  entry_main --> urn_noocodec_dag_react-agent-routing/node/scatter-chunks
  urn_noocodec_dag_react-agent-routing/node/scatter-chunks[/"scatter-chunks"/]
  urn_noocodec_dag_react-agent-routing/node/scatter-chunks -->|all-success| urn_noocodec_dag_react-agent-routing/node/collect-chunks
  urn_noocodec_dag_react-agent-routing/node/scatter-chunks -->|partial| urn_noocodec_dag_react-agent-routing/node/collect-chunks
  urn_noocodec_dag_react-agent-routing/node/scatter-chunks -->|all-error| urn_noocodec_dag_react-agent-routing/node/collect-chunks
  urn_noocodec_dag_react-agent-routing/node/scatter-chunks -->|empty| urn_noocodec_dag_react-agent-routing/node/end
  urn_noocodec_dag_react-agent-routing/node/collect-chunks{"collect-chunks"}
  urn_noocodec_dag_react-agent-routing/node/collect-chunks -->|success| urn_noocodec_dag_react-agent-routing/node/end
  urn_noocodec_dag_react-agent-routing/node/collect-chunks -->|error| urn_noocodec_dag_react-agent-routing/node/end
  urn_noocodec_dag_react-agent-routing/node/collect-chunks -->|empty| urn_noocodec_dag_react-agent-routing/node/end
  urn_noocodec_dag_react-agent-routing/node/end((("end")))

• Conversational Agents - the 8-node agent loop authored as JSON-LD
• Chat Event Orchestration - host one registered agent DAG behind EventTrigger or RequestTrigger
• Streaming Producers - StreamChannel, DagStreamProducer, and the scatter-source idiom this guide reuses
• Example: ReAct agent memory - working example: trace streaming, live token deltas, provenance recall
• Example: ReAct agent routing - working example: one shared sink demultiplexes two concurrent conversations by routeKey

What It Lets You Do

Use when

Use this guide when mapping a ReAct-style agent onto Dagonizer's DAG primitives. It is for applications that need reasoning traces, model token streaming, tool dispatch, provenance, recall, or concurrent stream routing without leaving the JSON-LD graph model.

Code Samples

The sections below explain how the runnable ReAct examples hang together: the loop mapping, trace streaming, live token deltas, RDF provenance, and stream demultiplexing.

Details for Nerds

The 8-node loop IS ReAct

ReAct moment	Node	What it produces
Thought	call-model	The model's reply — either a plan to call a tool, or a direct answer.
Action	decode-tools	The tool call decoded from the model's response (name + arguments).
Observation	collect-results	The gathered tool output, folded back after the scatter dispatch.
Final	append-assistant	The terminal text answer; the loop does not restart.

The loop-back edge — collect-results → build-request — is the "Act, then reason again" step: after tools run, the request is rebuilt with the tool observation in history and the model is called again. append-assistant routes to end-done (completed); no loop-back follows a Final step. Nothing about this mapping is metaphorical: it names the exact same eight registered nodes and edges documented in Conversational Agents § Agent loop.

Streaming the reasoning trace

A running agent loop already emits one NodeResultType per node per turn — Dagonizer.execute(dagIri, state) returns an Execution, which is both an AsyncIterable<NodeResultType<NodeStateInterface>> and a PromiseLike. To observe the ReAct trace live, organize it as a stream using the framework from Streaming Producers: no new mechanism, the same DagStreamProducer → StreamChannel.driven → outer scatter idiom used everywhere else in the docs.

AgentTraceProducer (exported from @studnicky/dagonizer/patterns) is a DagStreamProducer<ReasoningTraceItemType> purpose-built for this. Each emitted item pairs a ReasoningStepType with a monotonic ordinal assigned at emission time — so a downstream recorder can derive a wasInformedBy-style chain from item.ordinal - 1 alone, with no cross-item state and no dependence on the order items are actually recorded in. Subclass it:

import { AgentTraceProducer } from '@studnicky/dagonizer/patterns';
import type { NodeResultType } from '@studnicky/dagonizer';
import type { NodeStateInterface } from '@studnicky/dagonizer';

class ReActTraceProducer extends AgentTraceProducer {
  protected describe(stage: NodeResultType<NodeStateInterface>): string {
    // Map the stage's state back to a human-readable string for this step.
    // AgentTraceProducer already decides WHICH kind (thought/action/
    // observation/final) the stage represents from its fixed node-name map.
    return stage.nodeName; // replace with your state-shaped extraction
  }
}

The constructor takes the running loop's Execution directly:

const execution = agentDispatcher.execute('urn:noocodec:dag:react-agent', agentState);
const traceState = new TraceState();
traceState.source = StreamChannel.driven(new ReActTraceProducer(execution));

AgentTraceProducer maps call-model → thought, decode-tools → action, collect-results → observation, append-assistant → final — the same four rows from the table above. select(stage) is fully implemented on the base class; you only implement describe(stage).

The outer DAG is an ordinary scatter over traceState.source, with a body node that records or renders each ReasoningTraceItemType as it arrives:

ReActTraceProducer (DagStreamProducer<ReasoningTraceItemType>)
  → StreamChannel.driven(producer)
    → traceState.source
      → outer ScatterNode (body: your per-step node)

Because every item carries its own ordinal, the outer scatter's concurrency is a free performance choice, not a correctness constraint — raising it above 1 does not corrupt the chain.

Draining the outer scatter is what pulls the inner agent loop forward — the same back-pressure contract every DagStreamProducer reader relies on.

Live token streaming

CallModelNode accepts a { sink } option in its constructor and forwards it to adapter.chatStream(request, sink) instead of the buffered adapter.chat(request):

class MyCallModelNode extends CallModelNode<AgentState> {
  readonly name = 'call-model';
  constructor(llm: LlmAdapterInterface, options: { sink?: StreamSinkInterface<ChatStreamChunkType> } = {}) {
    super(llm, options);
  }
  // … getRequest / storeResponse …
}

const sink: StreamSinkInterface<ChatStreamChunkType> = {
  async push(chunk) { process.stdout.write(chunk.delta); },
};
new MyCallModelNode(llm, { sink });

ChatStreamChunkType carries one field, delta — the text fragment produced since the previous chunk. Every LlmAdapterInterface implements chatStream(request, sink): Promise<ChatResponseType>:

• BaseAdapter's default performChatStream is buffered — it runs the same guarded/classified/retried path as chat(), then pushes one chunk carrying the full response text. An adapter that never overrides performChatStream still works with { sink }; it just never streams more than one chunk.
• The anthropic, gemini-api, gemini-nano, web-llm, and ollama adapters override performChatStream with real provider token streaming — the sink receives one push per token/fragment as the provider emits it.
• Tool-turn responses (variant: 'tools' | 'mixed') use a buffered push in every adapter — there is no per-token text to stream when the model is emitting a structured tool call.

No other node or wiring changes; the sink is purely an observation channel alongside the returned ChatResponseType, which storeResponse still writes to state exactly as it does without streaming.

Capturing thoughts to a graph with provenance, and recalling them

Each ReasoningTraceItemType scattered by the outer DAG can be asserted into a triple store as a dag:Reasoning-shaped provenance record, chained with wasGeneratedBy (links a step to the run that produced it) and wasInformedBy (links a step to the item at item.ordinal - 1 in the same run). A body node like RecordReasoningStepNode in the example writes these quads per scattered item, deriving the subject IRI and the wasInformedBy link purely from item.ordinal — no instance state, correct at any scatter concurrency.

A later run recalls prior reasoning by walking the same graph: find the prior run's final step, follow wasInformedBy backward to the first step, and format the chain as a one-line hint. That hint is injected as a leading system message on the next run's build-request, so the model sees what it concluded last time before reasoning again.

This is exactly the pattern the Archivist uses in production: its recall-context node walks prior dag:Reasoning provenance quads to inform new decisions, and its recording node asserts the same wasGeneratedBy / wasInformedBy chain documented here.

See Example: ReAct agent memory for the complete, runnable version of trace streaming, live token deltas, provenance recording, and cross-run recall.

Routing concurrent streams — the sink is a DAG

The { sink } option on CallModelNode is bound once, per node INSTANCE — not per execution. That is deliberate: CallModelNode.execute wraps this.sink in fresh RoutingStreamSink instances for the batch items it processes, via RoutingStreamSink.of(this.sink, this.routeKey(state), source). Each pushed ChatStreamChunkType ({delta}) becomes a self-describing RoutedChatStreamChunkType at the downstream sink — {routeKey, delta, source} — where source is {dagName, nodeName} read from the executing NodeContextType. One shared node instance, streaming into one shared sink, is therefore safe for many CONCURRENT runs: every chunk already carries enough information for a downstream subscriber to tell whose run it belongs to. The dagName property is provenance metadata from the execution context; the registered DAG identity still comes from the DAG IRI.

routeKey(state) is the seam that supplies the demultiplexing key. The default returns '' (a single unrouted stream, the case covered above). Override it to read a per-run id from state:

class RoutingCallModelNode extends MyCallModelNode {
  protected override routeKey(state: AgentState): string {
    return state.conversationId;
  }
}

Because the routing key comes from state, not from the node instance, ONE RoutingCallModelNode registered on ONE dispatcher correctly demultiplexes as many concurrent dispatcher.execute('urn:noocodec:dag:react-agent', state) calls as are run in parallel — each state carries its own conversationId.

The sink itself does not need to be a passive buffer. Because RoutedChatStreamChunkType is a JSON Schema-derived entity like any other, a StreamChannel<RoutedChatStreamChunkType> — the same channel type used for streaming producers — can be handed to CallModelNode as the shared sink, and ALSO passed as the source of an ordinary ScatterNode that classifies each chunk by routeKey and routes it into a per-conversation destination (here, a TranscriptStore.append(routeKey, delta)):

channel: StreamChannel<RoutedChatStreamChunkType>   ← shared sink AND shared scatter source
  → RoutingCallModelNode (one instance, { sink: channel })
      conversation 'c1' → chatStream(...) → RoutingStreamSink → channel.push({routeKey:'c1', delta, source})
      conversation 'c2' → chatStream(...) → RoutingStreamSink → channel.push({routeKey:'c2', delta, source})
  → routing DAG: ScatterNode over channel
      → body: RouteChunkNode — appends item.delta into TranscriptStore[item.routeKey]

This is the same DagStreamProducer → StreamChannel → outer-scatter idiom used for the reasoning-trace stream above; the difference is that the source here is a live, concurrently-fed sink rather than a single producer's output, and the scatter body classifies each item by a field on its own payload (routeKey) instead of accumulating everything into one bucket.

The runnable Example: ReAct agent routing starts the routing DAG's drain first (so the channel's bounded buffer never backs up two conversations' pushes), runs two conversations concurrently against the one shared node and one shared channel, closes the channel once both conversations finish, then awaits the routing drain — reconstructing each conversation's transcript separately from the interleaved chunk stream.

Related Concepts

• Conversational Agents - the 8-node agent loop authored as JSON-LD
• Streaming Producers - StreamChannel, DagStreamProducer, and the scatter-source idiom this guide reuses
• Example: ReAct agent memory - working example: trace streaming, live token deltas, provenance recall
• Example: ReAct agent routing - working example: one shared sink demultiplexes two concurrent conversations by routeKey
• Example 29: Agent DAG - the 8-node JSON-LD topology this guide annotates as ReAct