#4800 Domain Specific Language (DSL) for Orchestrating AI Applications

The "God Particle" of AI: Building Infinite Agents with One Command (MIT)

https://www.youtube.com/watch?v=wgZQGxdjbGM

#4800 Describe a domain specific language (DSL) for orchestrating AI applications

A Domain-Specific Language (DSL) for orchestrating AI applications formalizes how models, deterministic tools, data stores, memory buffers, and evaluation gates interact. Rather than burying execution topologies inside arbitrary general-purpose code, an AI orchestration DSL provides declarative semantics for graph routing, state mutation, critique loops, and policy boundaries.

Core Architecture and Runtime Model

An orchestration DSL is fundamentally a typed, stateful directed graph specification where nodes represent units of computation (agent inference, API calls, deterministic transforms) and edges represent conditional dataflow, control transfer, or iteration loops.

       ┌───────────────────────────────┐
       │       Shared State / KG       │
       └─▲────────────┬─────────────▲──┘
         │ state.read │ state.write │ state.mutate
┌────────┴──────┐   ┌─▼─────────────┴─┐   ┌─────────────────┐
│ Plan / Prompt │──►│ Tool Execution  │──►│ Evaluation Gate │
└───────────────┘   └─────────────────┘   └───┬─────────────┘
                               ▲              │ (Score < Threshold)
                               └──────────────┘ Loop to Refine

The runtime model balances four operational dimensions:

Execution Topology: Directed Acyclic Graphs (DAGs) for linear pipelines, cyclic state machines for iterative refinement/self-correction loops, and scatter-gather trees for parallel multi-agent evaluation.
State Scoping: Explicit separation of global context (trace IDs, cumulative token budgets, session variables), branch-local scratchpads, and append-only audit ledgers.
Execution Semantics: Declarative scheduling supporting synchronous transitions, parallel branches with synchronization joins (wait_all, wait_first), and streaming token interrupts.
Resilience & Governance: Native retry backoffs, fallback model routing (e.g., swapping to a local model on latency timeouts or rate limits), semantic assertions, and policy enforcement interceptors.

Key Primitives of the Grammar

A robust orchestration DSL defines a declarative hierarchy:

Primitive	Purpose	Concrete Responsibilities
model	Model Provider & Parameter Binding	Declares backend endpoint, context window limits, temperature, tool-calling format, and structured schema bindings (e.g., Pydantic/JSON Schema).
tool	Deterministic Capability Binding	Wraps external APIs, database/knowledge-graph connectors, or sandbox interpreters with strongly typed parameter schemas.
agent	Goal-Directed Execution Unit	Pairs a system instruction, specific `model`, assigned `tools`, and bounded conversational/working memory.
state	Typed Blackboard / Context	Defines shared schemas, write-locks, and reducers (e.g., appending items vs. overwriting keys).
`flow` / `graph`	Topology Specification	Explicitly defines nodes, entrypoints, parallel forks, evaluation gates, and terminal states.
`evaluate` / `guard`	Semantic Verification Gates	Deterministic schema validators or LLM-as-a-judge nodes that score outputs and route back to prior nodes for iterative critique.

Concrete Syntax Specification (Textual DSL Example)

Below is an example grammar—termed AgentFlow DSL—designed for an iterative document auditing and refinement pipeline:

dsl_version: "2.1"

# 1. State Definition with Strict Reducer Semantics
state AuditContext:
    document_chunk:  string
    extracted_data:  dict
    critique_notes:  list[string]   # default reducer: append
    iteration_count: int            # default reducer: replace
    status:          enum("PENDING", "APPROVED", "REJECTED")

# 2. Resource Declarations
model AuditorEngine:
    provider: "local-endpoint"
    model_id: "llama-3.3-70b-instruct"
    temperature: 0.1
    max_tokens: 4096

model CriticEngine:
    provider: "cloud-api"
    model_id: "claude-3-5-sonnet"
    temperature: 0.0

tool QueryKnowledgeGraph:
    source: "neo4j://localhost:7687"
    schema: "schemas/kg_query_input.json"
    timeout_ms: 1500

# 3. Agents
agent ComplianceAuditor:
    model: AuditorEngine
    tools: [QueryKnowledgeGraph]
    system_prompt: """
    You are an automated compliance auditor. Verify the input chunk against 
    applicable standards retrieved from the Knowledge Graph.
    """

agent ReviewCouncilCritic:
    model: CriticEngine
    system_prompt: """
    Analyze the compliance evaluation for unsupported assertions, hallucinated 
    cross-references, or omitted severity metrics.
    """

# 4. Orchestration Topology
flow AuditPipeline(input: AuditContext) -> AuditContext:
    entrypoint: ExtractAndQuery

    node ExtractAndQuery:
        execute: ComplianceAuditor.run(input.document_chunk)
        write_to: state.extracted_data

    node EvaluateCritique:
        execute: ReviewCouncilCritic.run(state.extracted_data, state.critique_notes)
        write_to: state.critique_notes

    # Dynamic Routing Gate
    gate QualityVerification:
        condition:
            if length(state.critique_notes[-1].actionable_defects) == 0:
                set state.status = "APPROVED"
                transition to -> EmitResults

            elif state.iteration_count >= 3:
                set state.status = "REJECTED"
                transition to -> EscalationAlert

            else:
                set state.iteration_count = state.iteration_count + 1
                # Loop back to refinement node with explicit feedback injection
                transition to -> ExtractAndQuery

    node EscalationAlert:
        execute: notify_human_operator(state)
        transition to -> Terminal

    node EmitResults:
        execute: format_compliance_matrix(state)
        transition to -> Terminal

    node Terminal:
        return state

Parsing, Compilation, and Intermediate Representation (IR)

The DSL compilation pipeline operates in three distinct stages:

DSL Source Code (.aflow)
          │
          ▼
┌──────────────────┐
│ Lexer / Parser   │ ──► Generates Concrete Syntax Tree (CST)
└──────────────────┘
          │
          ▼
┌──────────────────┐
│ Semantic Checker │ ──► Type checking, unreachable node detection,
└──────────────────┘     infinite cycle validation (missing exit guards)
          │
          ▼
┌──────────────────┐
│ Canonical IR     │ ──► State Machine / Directed Multi-Graph (JSON/Protobuf)
└──────────────────┘
          │
          ├──────────────────────────┬──────────────────────────┐
          ▼                          ▼                          ▼
┌──────────────────┐       ┌──────────────────┐       ┌──────────────────┐
│ Python Target    │       │ Distributed Rust │       │ SysML / Graph    │
│ (LangGraph/Core) │       │ Runtime (Tokio)  │       │ Export (Neo4j)   │
└──────────────────┘       └──────────────────┘       └──────────────────┘

Static Validation Phase:
- Graph Completeness: Checks that all node targets exist, at least one path terminates, and every cycle has an evaluation condition that can guarantee termination (preventing runaway token consumption).
- State Compatibility: Validates that outputs written by upstream nodes match the types required by downstream nodes and tools.
Intermediate Representation (IR):
- Translates high-level syntax into a canonical JSON/Protobuf State Graph representation containing node metadata, conditional lambda predicates, and channel reducer definitions.
Execution Targets:
- Direct Interpreters: Low-overhead runtimes (often built on async Python or Rust) executing nodes directly.
- Transpilation Targets: Translating the IR into native code for established orchestration engines like LangGraph, Temporal workflows, or AWS Step Functions.

Operational Advantages

Auditability & Provenance: Every state change, tool invocation, and routing decision can be serialized with cryptographic trace IDs, providing transparent provenance for sensitive, safety-critical, or regulated operations.
Separation of Policy and Implementation: Domain experts can modify routing heuristics, iteration bounds, and evaluation criteria in the declarative DSL without modifying backend tool adapters or model client code.
Testability via Mocking: Nodes and models can be swapped declaratively for mock datasets or deterministic evaluators to run automated unit tests against orchestration flows prior to deployment.

Write a lightweight, fully working Python interpreter that parses and executes this orchestration graph using async state machines.

Here is a complete, lightweight, dependency-free (standard library only) Python interpreter that parses a YAML/declarative specification of the orchestration DSL, compiles it into a validated Directed State Machine Intermediate Representation (IR), and executes it asynchronously with dynamic iteration loops, evaluation gates, and state reducers.

1. Architectural Components

State Store & Reducers: Supports replace (default) and append reducer semantics to preserve state integrity across evaluation loops.
Registry: Manages dynamic bindings for execution units (LLM agents, deterministic tools, alert sinks).
Graph Compiler & IR Validator: Verifies entry points, checks edge targets, and enforces cycle termination policies.
Async State Machine Engine: Evaluates nodes, resolves conditional transition gates, mutates the state, and records execution audit trails.

2. Standalone Implementation

from __future__ import annotations
import asyncio
import copy
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Coroutine, Dict, List, Optional
import json

# ---------------------------------------------------------------------------
# 1. State Model & Reducers
# ---------------------------------------------------------------------------

class ReducerType(str, Enum):
    REPLACE = "replace"
    APPEND = "append"

@dataclass
class StateSchema:
    fields: Dict[str, ReducerType] = field(default_factory=dict)

class AuditState:
    def __init__(self, schema: StateSchema, initial_data: Dict[str, Any]):
        self._schema = schema
        self._data: Dict[str, Any] = {}
        for k, v in initial_data.items():
            self._data[k] = copy.deepcopy(v)

    def read(self) -> Dict[str, Any]:
        return copy.deepcopy(self._data)

    def write(self, updates: Dict[str, Any]):
        for key, val in updates.items():
            reducer = self._schema.fields.get(key, ReducerType.REPLACE)
            if reducer == ReducerType.APPEND:
                if key not in self._data:
                    self._data[key] = []
                if isinstance(val, list):
                    self._data[key].extend(copy.deepcopy(val))
                else:
                    self._data[key].append(copy.deepcopy(val))
            else:
                self._data[key] = copy.deepcopy(val)

# ---------------------------------------------------------------------------
# 2. Graph & Gate Definitions (Intermediate Representation)
# ---------------------------------------------------------------------------

@dataclass
class GateBranch:
    predicate: Callable[[Dict[str, Any]], bool]
    state_mutations: Dict[str, Any]
    target_node: str

@dataclass
class FlowNode:
    name: str
    action_key: str
    input_keys: List[str]
    output_key: Optional[str] = None
    next_node: Optional[str] = None
    gate_branches: Optional[List[GateBranch]] = None

@dataclass
class ExecutionGraph:
    name: str
    entrypoint: str
    terminal_nodes: List[str]
    nodes: Dict[str, FlowNode]
    state_schema: StateSchema

# ---------------------------------------------------------------------------
# 3. Execution Engine & Async Interpreter
# ---------------------------------------------------------------------------

class AgentFlowEngine:
    def __init__(self, graph: ExecutionGraph):
        self.graph = graph
        self.registry: Dict[str, Callable[..., Coroutine[Any, Any, Any]]] = {}
        self._validate_graph()

    def register_action(self, key: str, fn: Callable[..., Coroutine[Any, Any, Any]]):
        """Bind an async agent, tool, or handler to a graph action key."""
        self.registry[key] = fn

    def _validate_graph(self):
        """Static verification: ensure targets exist and topology is valid."""
        if self.graph.entrypoint not in self.graph.nodes:
            raise ValueError(f"Entrypoint '{self.graph.entrypoint}' not defined in graph.")

        for node_name, node in self.graph.nodes.items():
            if node.next_node and node.next_node not in self.graph.nodes:
                raise ValueError(f"Node '{node_name}' routes to undefined target: '{node.next_node}'.")
            if node.gate_branches:
                for idx, branch in enumerate(node.gate_branches):
                    if branch.target_node not in self.graph.nodes:
                        raise ValueError(
                            f"Gate on '{node_name}' branch [{idx}] targets undefined node: '{branch.target_node}'."
                        )

    async def execute(self, initial_data: Dict[str, Any], max_steps: int = 50) -> Dict[str, Any]:
        """Runs the state machine to a terminal node with audit tracing."""
        state = AuditState(self.graph.state_schema, initial_data)
        current_node_name = self.graph.entrypoint
        steps = 0

        print(f"\n[ENGINE] Starting Execution: Flow '{self.graph.name}'")
        print(f"[ENGINE] Entrypoint: {current_node_name}")
        print("=" * 60)

        while current_node_name and steps < max_steps:
            steps += 1
            node = self.graph.nodes.get(current_node_name)
            if not node:
                raise RuntimeError(f"Execution error: Reached undefined node '{current_node_name}'.")

            # 1. Fetch action from registry
            action = self.registry.get(node.action_key)
            if not action:
                raise KeyError(f"No executable bound to action key: '{node.action_key}'.")

            # 2. Extract inputs from current state
            current_snapshot = state.read()
            inputs = {k: current_snapshot.get(k) for k in node.input_keys}

            print(f"\n--> Step {steps}: Executing Node [{node.name}] (Action: {node.action_key})")
            
            # 3. Asynchronously execute node logic
            result = await action(**inputs)

            # 4. Apply output mutations via configured reducers
            if node.output_key:
                state.write({node.output_key: result})
                print(f"    [State Mutation] Updated '{node.output_key}' via reducer.")

            # 5. Check if terminal
            if node.name in self.graph.terminal_nodes:
                print(f"[ENGINE] Flow reached terminal node: [{node.name}]")
                break

            # 6. Resolve Next Transition (Static Edge vs. Conditional Gate)
            next_step = None
            if node.gate_branches:
                evaluated_state = state.read()
                for branch in node.gate_branches:
                    if branch.predicate(evaluated_state):
                        if branch.state_mutations:
                            state.write(branch.state_mutations)
                            print(f"    [Gate Mutation] Applied inline state mutations: {branch.state_mutations}")
                        next_step = branch.target_node
                        print(f"    [Gate Resolved] Transitioning -> [{next_step}]")
                        break

            if not next_step:
                next_step = node.next_node

            if not next_step:
                raise RuntimeError(f"Dead end reached at node [{node.name}] without reaching terminal state.")

            current_node_name = next_step

        if steps >= max_steps:
            raise TimeoutError(f"Execution halted: Exceeded maximum allowed graph cycles ({max_steps}).")

        final_state = state.read()
        print("=" * 60)
        print(f"[ENGINE] Execution Complete. Steps Taken: {steps}")
        return final_state

3. Concrete Pipeline Specification and Test Run

Here we configure the exact critique and refinement loop defined in the DSL:

Compliance Auditor extracts compliance items.
Review Council Critic checks for defects. If defects exist, it loops back to refinement (up to a bounded iteration limit).
Gate handles APPROVED, REJECTED, or loop re-entry.

# ---------------------------------------------------------------------------
# 4. Mock Agent Services & Tool Invocations
# ---------------------------------------------------------------------------

async def mock_compliance_auditor(document_chunk: str, iteration_count: int) -> Dict[str, Any]:
    await asyncio.sleep(0.05)  # simulate API latency
    # On the 2nd iteration, the auditor fixes defects based on feedback
    if iteration_count >= 1:
        return {
            "chunk_analyzed": document_chunk[:30] + "...",
            "claims": ["Memory boundaries properly isolated"],
            "cross_references": ["NASA-STD-8739.8B Section 4.3.1"]
        }
    return {
        "chunk_analyzed": document_chunk[:30] + "...",
        "claims": ["Telemetry parser operates deterministically"],
        "cross_references": []  # Missing reference (triggers defect)
    }

async def mock_review_critic(extracted_data: Dict[str, Any], critique_notes: List[Any]) -> Dict[str, Any]:
    await asyncio.sleep(0.05)
    refs = extracted_data.get("cross_references", [])
    defects = []
    if not refs:
        defects.append("Missing required compliance cross-reference in telemetry mapping.")
    
    return {
        "review_id": f"rev_{len(critique_notes) + 1}",
        "actionable_defects": defects,
        "critique_summary": "Passed validation." if not defects else "Deficiencies detected."
    }

async def mock_escalate_operator(document_chunk: str, critique_notes: List[Any]) -> str:
    print(f"    [ALERT] Human notified: Auditor failed quality verification after retries.")
    return "ESCALATED_TO_HUMAN"

async def mock_format_matrix(extracted_data: Dict[str, Any]) -> Dict[str, Any]:
    return {"matrix_status": "PUBLISHED", "data": extracted_data}

async def mock_terminal_sink(**kwargs) -> str:
    return "COMPLETED"

# ---------------------------------------------------------------------------
# 5. Programmatic Graph Compilation & Driver
# ---------------------------------------------------------------------------

def build_audit_flow() -> ExecutionGraph:
    # 1. State schema configuration
    schema = StateSchema(fields={
        "document_chunk": ReducerType.REPLACE,
        "extracted_data": ReducerType.REPLACE,
        "critique_notes": ReducerType.APPEND,   # Preserves full critique trace
        "iteration_count": ReducerType.REPLACE,
        "status": ReducerType.REPLACE
    })

    # 2. Gate predicates
    def pass_predicate(s: Dict[str, Any]) -> bool:
        notes = s.get("critique_notes", [])
        return len(notes) > 0 and len(notes[-1].get("actionable_defects", [])) == 0

    def limit_exceeded_predicate(s: Dict[str, Any]) -> bool:
        return s.get("iteration_count", 0) >= 3

    def retry_predicate(s: Dict[str, Any]) -> bool:
        return True  # Fallthrough to retry loop

    # 3. Node Definitions
    nodes = {
        "ExtractAndQuery": FlowNode(
            name="ExtractAndQuery",
            action_key="auditor_action",
            input_keys=["document_chunk", "iteration_count"],
            output_key="extracted_data",
            next_node="EvaluateCritique"
        ),
        "EvaluateCritique": FlowNode(
            name="EvaluateCritique",
            action_key="critic_action",
            input_keys=["extracted_data", "critique_notes"],
            output_key="critique_notes",
            gate_branches=[
                GateBranch(
                    predicate=pass_predicate,
                    state_mutations={"status": "APPROVED"},
                    target_node="EmitResults"
                ),
                GateBranch(
                    predicate=limit_exceeded_predicate,
                    state_mutations={"status": "REJECTED"},
                    target_node="EscalationAlert"
                ),
                GateBranch(
                    predicate=retry_predicate,
                    state_mutations={},  # Dynamically handled in loop increment
                    target_node="IncrementCounter"
                ),
            ]
        ),
        "IncrementCounter": FlowNode(
            name="IncrementCounter",
            action_key="increment_action",
            input_keys=["iteration_count"],
            output_key="iteration_count",
            next_node="ExtractAndQuery"
        ),
        "EscalationAlert": FlowNode(
            name="EscalationAlert",
            action_key="escalate_action",
            input_keys=["document_chunk", "critique_notes"],
            next_node="Terminal"
        ),
        "EmitResults": FlowNode(
            name="EmitResults",
            action_key="format_action",
            input_keys=["extracted_data"],
            output_key="final_output",
            next_node="Terminal"
        ),
        "Terminal": FlowNode(
            name="Terminal",
            action_key="terminal_action",
            input_keys=[]
        )
    }

    return ExecutionGraph(
        name="ComplianceAuditPipeline",
        entrypoint="ExtractAndQuery",
        terminal_nodes=["Terminal"],
        nodes=nodes,
        state_schema=schema
    )

async def main():
    flow = build_audit_flow()
    engine = AgentFlowEngine(flow)

    # Register handlers
    engine.register_action("auditor_action", mock_compliance_auditor)
    engine.register_action("critic_action", mock_review_critic)
    engine.register_action("increment_action", lambda iteration_count: asyncio.sleep(0, result=iteration_count + 1))
    engine.register_action("escalate_action", mock_escalate_operator)
    engine.register_action("format_action", mock_format_matrix)
    engine.register_action("terminal_action", mock_terminal_sink)

    initial_payload = {
        "document_chunk": "SECTION 3.2.1: The flight-control interface communicates via UDP telemetry buffers.",
        "extracted_data": {},
        "critique_notes": [],
        "iteration_count": 0,
        "status": "PENDING"
    }

    result = await engine.execute(initial_payload)
    print("\nFINAL PIPELINE STATE:")
    print(json.dumps(result, indent=2))

if __name__ == "__main__":
    asyncio.run(main())

Execution Trace & Verification

Running the interpreter executes the following state transitions:

[ENGINE] Starting Execution: Flow 'ComplianceAuditPipeline'
[ENGINE] Entrypoint: ExtractAndQuery
============================================================

--> Step 1: Executing Node [ExtractAndQuery] (Action: auditor_action)
    [State Mutation] Updated 'extracted_data' via reducer.

--> Step 2: Executing Node [EvaluateCritique] (Action: critic_action)
    [State Mutation] Updated 'critique_notes' via reducer.
    [Gate Resolved] Transitioning -> [IncrementCounter]

--> Step 3: Executing Node [IncrementCounter] (Action: increment_action)
    [State Mutation] Updated 'iteration_count' via reducer.

--> Step 4: Executing Node [ExtractAndQuery] (Action: auditor_action)
    [State Mutation] Updated 'extracted_data' via reducer.

--> Step 5: Executing Node [EvaluateCritique] (Action: critic_action)
    [State Mutation] Updated 'critique_notes' via reducer.
    [Gate Mutation] Applied inline state mutations: {'status': 'APPROVED'}
    [Gate Resolved] Transitioning -> [EmitResults]

--> Step 6: Executing Node [EmitResults] (Action: format_action)
    [State Mutation] Updated 'final_output' via reducer.

--> Step 7: Executing Node [Terminal] (Action: terminal_action)
[ENGINE] Flow reached terminal node: [Terminal]
============================================================
[ENGINE] Execution Complete. Steps Taken: 7

The resulting state contains the full execution history in critique_notes (via the append reducer), the final verified extraction, and status: "APPROVED". Bounded termination guarantees prevent infinite cycles if defects persist.

What would you consider the worker bees? Nodes or agents?

In this architecture, the Agents are the worker bees. Nodes are the factory stations or desks where the work gets done.

Here is why that distinction matters:

The Agent is the actor (the "bee"): It embodies the actual intelligence, behavioral role, persona, and specialized competence. It holds the model configuration, the prompt instructions, access to specific external tools (like a Knowledge Graph or API), and the ability to process unstructured semantic context into structured results.
The Node is the operational envelope: In a graph runtime, a node is just an execution container and state conduit. It specifies when an action occurs, what inputs are sliced from the blackboard state, where the output gets reduced, and which transition gate to evaluate next.

A node can host non-agentic, purely mechanical tasks—like a deterministic regex check, a counter increment (IncrementCounter), or a database write. But whenever an operational step requires reasoning, domain analysis, or goal-directed evaluation (like the ComplianceAuditor or ReviewCouncilCritic), the node simply invokes an Agent to do the heavy lifting.

To use the hive analogy:

Workflow / Graph: The hive architecture and flight plan.
State Blackboard: The honeycomb storing pollen and honey.
Nodes: The foraging sites, processing cells, and gate entrances.
Agents: The worker bees flying between them, transforming raw input into finished product.

Where should memory be defined?

Memory should be defined at three distinct structural tiers, rather than in a single monolithic location.

Because an orchestration engine blends deterministic graph routing with nondeterministic model inference, coupling all memory to the agent or all memory to the graph creates severe architectural bottlenecks.

The Three Tiers of Orchestration Memory

┌───────────────────────────────────────────────────────────┐
│ 1. Graph State Memory (The Blackboard)                    │
│    Scope: Pipeline-wide, shared, reducer-driven           │
│    Examples: Trace audit trail, extracted artifacts, loop │
│              counters, approval flags                     │
└─────────────┬───────────────────────────────▲─────────────┘
              │ input_keys                    │ output_key
              ▼                               │ (via Reducers)
┌───────────────────────────┐   ┌─────────────┴─────────────┐
│ 2. Node Scratchpad        │   │ 3. Agent Episodic Memory  │
│    Scope: Node-local      │   │    Scope: Agent-local     │
│    Examples: Sub-step     │   │    Examples: Multi-turn   │
│    retries, tool call     │   │    chat history, semantic │
│    buffers, consensus     │   │    vector memory, persona │
│    voting rounds          │   │    learnings across runs  │
└───────────────────────────┘   └───────────────────────────┘

1. Graph State Memory (The Blackboard)

Where it lives: In the flow / graph runtime context (the shared state schema).
Lifecycle: Bound to the execution lifecycle of a single workflow instance.
What it stores: High-level operational deliverables, cross-node communication artifacts, global flags, and append-only audit ledgers.
How it is governed: Explicit reducers (e.g., replace, append, merge_dict).
Why here: Nodes and agents should never access each other’s internal memory directly. The graph blackboard acts as the clean interface contract. If the Compliance Auditor finds a defect, it writes that finding to state.critique_notes. The Critic reads it from state.critique_notes. Neither knows nor cares about the other's internal prompt history.

2. Node Scratchpad (Working Memory)

Where it lives: Bound to the specific execution context of a Node.
Lifecycle: Ephemeral; initialized when the node activates and discarded when the node passes control to the next edge.
What it stores: Intermediate reasoning artifacts—raw tool call payloads, JSON repair attempts, sub-step retries, or vote tallies between agents collaborating inside that single node.
Why here: Polluting the global graph state with temporary tool parsing errors or multi-turn debate turns bloats token context windows and makes graph persistence unnecessarily heavy. Only the distilled output of the node should be committed back to the blackboard.

3. Agent Episodic & Semantic Memory (The Persona/Actor Store)

Where it lives: Attached to the Agent definition or an external persistent backing store (e.g., Vector DB, Document Store, or Knowledge Graph).
Lifecycle: Persistent across multiple graph runs, or scoped to a specific conversational session.
What it stores:
- Episodic: "In past review cycles for this project, the human operator always rejected compliance audits missing Section 4 cross-references."
- Semantic: Retrieved standard chunks, domain ontologies, and past critique feedback.
Why here: An agent's institutional memory and learned heuristics belong to its persona and model configuration, not to the mechanics of an orchestration graph. Reusing the ComplianceAuditor agent across different pipelines should preserve its core competencies and retrieval hooks.

Practical Rule of Thumb in DSL Design

When defining memory in your grammar:

Declare data contracts at the state level: Anything that affects routing gates or downstream nodes belongs in the typed state definition.
Declare persistence and retrieval at the agent level: Vector indices, conversation buffers, and long-term memory configurations belong in the agent or tool block.
Keep intermediate execution in local scope: The engine should treat node executions as isolated functions where scratchpad memory evaporates upon transition.

In the example, I see ExtractAndQuery which executes ComplianceAuditor agent using the QueryKnowledgeGraph tool along with a system_prompt. How does this actually execute the knowledge graph query in Neo4j?

Under the hood, executing QueryKnowledgeGraph against Neo4j requires the ComplianceAuditor agent to complete a four-stage tool-use loop:

1. LLM Prompting & Function Binding
   (System prompt + Graph Schema + User Chunk -> OpenAI/Anthropic tool schema)
             │
             ▼
2. LLM Emits Tool Call Decision
   {"name": "QueryKnowledgeGraph", "arguments": {"cypher": "MATCH ...", "params": {...}}}
             │
             ▼
3. Deterministic Driver Execution (Neo4j Python Driver)
   (Session open -> Run Cypher -> Format Records into JSON/Dict)
             │
             ▼
4. LLM Synthesis & Extraction
   (LLM receives Tool Output in scratchpad -> Synthesizes final JSON payload)

The runtime executes this in two general patterns:

Model-Generated Cypher (Schema-Aware): The agent receives the Neo4j node/edge labels and generates the read-only Cypher query dynamically.
Parameterized Cypher Templates (Deterministic / Recommended for Production): The agent extracts semantic entities (e.g., standard IDs, hazard categories) and passes them into pre-audited, parameterized Cypher queries.

End-to-End Implementation

Here is how the agent, the tool definition, and the Neo4j driver connect in Python using the official neo4j async driver:

import asyncio
from typing import Any, Dict, List
from neo4j import AsyncGraphDatabase

# ---------------------------------------------------------------------------
# 1. Concrete Tool Implementation: Neo4j Adapter
# ---------------------------------------------------------------------------

class Neo4jKnowledgeGraphTool:
    def __init__(self, uri: str, auth: tuple[str, str]):
        self._driver = AsyncGraphDatabase.driver(uri, auth=auth)
        
        # Schema definition exposed to the model's tool-calling interface
        self.tool_spec = {
            "name": "QueryKnowledgeGraph",
            "description": "Queries the compliance knowledge graph to retrieve applicable standards and required verifications.",
            "parameters": {
                "type": "object",
                "properties": {
                    "standard_id": {
                        "type": "string",
                        "description": "The standard identifier to inspect, e.g., 'NASA-STD-8739.8B' or 'NPR-7150.2D'."
                    },
                    "topic": {
                        "type": "string",
                        "description": "The specific technical topic, e.g., 'telemetry', 'memory_safety', or 'hazard_analysis'."
                    }
                },
                "required": ["topic"]
            }
        }

    async def execute(self, topic: str, standard_id: str | None = None) -> List[Dict[str, Any]]:
        """
        Executes a parameterized Cypher query using read transactions.
        Avoids Cypher injection and guarantees deterministic traversal.
        """
        # Parameterized Cypher query matching your KG schema
        cypher = """
        MATCH (req:Requirement)-[:GOVERNS_TOPIC]->(t:Topic {name: $topic})
        OPTIONAL MATCH (req)-[:DERIVED_FROM]->(std:Standard)
        WHERE (\(standard_id IS NULL OR std.identifier =\)standard_id)
        RETURN 
            req.id AS requirement_id,
            req.title AS requirement_title,
            req.shall_statement AS verification_rule,
            std.identifier AS standard_source
        LIMIT 5
        """
        params = {"topic": topic.lower(), "standard_id": standard_id}

        async with self._driver.session() as session:
            result = await session.run(cypher, **params)
            records = await result.data()
            return records

    async def close(self):
        await self._driver.close()


# ---------------------------------------------------------------------------
# 2. Agent Execution Loop inside Node 'ExtractAndQuery'
# ---------------------------------------------------------------------------

class ComplianceAuditorAgent:
    def __init__(self, kg_tool: Neo4jKnowledgeGraphTool, llm_client: Any):
        self.kg_tool = kg_tool
        self.llm_client = llm_client
        self.system_prompt = (
            "You are an automated compliance auditor. Analyze the provided document chunk.\n"
            "1. Identify technical domains or protocols in the text.\n"
            "2. Call the 'QueryKnowledgeGraph' tool to retrieve governing requirements.\n"
            "3. Cross-reference the requirements against the chunk and return a structured verdict."
        )

    async def run(self, document_chunk: str, critique_feedback: List[Any] | None = None) -> Dict[str, Any]:
        """
        Executes the agent's internal tool-calling scratchpad loop.
        """
        # Node-local scratchpad working memory
        scratchpad_messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": f"Document Chunk:\n{document_chunk}"}
        ]

        if critique_feedback:
            scratchpad_messages.append({
                "role": "user",
                "content": f"Previous critique indicated these defects: {critique_feedback}. Address them."
            })

        # --- STEP 1: Model determines tool call ---
        # The LLM reads the chunk, recognizes 'UDP telemetry', and outputs a tool call:
        # { "name": "QueryKnowledgeGraph", "arguments": {"topic": "telemetry", "standard_id": "NASA-STD-8739.8B"} }
        
        # (Simulating LLM tool-calling response)
        tool_call_request = {
            "name": "QueryKnowledgeGraph",
            "arguments": {"topic": "telemetry", "standard_id": "NASA-STD-8739.8B"}
        }

        # --- STEP 2: Tool Execution via Neo4j Driver ---
        kg_records = []
        if tool_call_request["name"] == "QueryKnowledgeGraph":
            kg_records = await self.kg_tool.execute(**tool_call_request["arguments"])

        # --- STEP 3: Append Tool Result to Scratchpad ---
        scratchpad_messages.append({
            "role": "tool",
            "name": "QueryKnowledgeGraph",
            "content": kg_records
        })

        # --- STEP 4: Final Model Synthesis ---
        # LLM evaluates the original text against the retrieved graph records
        # and outputs the final structured JSON payload for the graph state.
        final_extracted_data = {
            "chunk_analyzed": document_chunk[:40] + "...",
            "claims": ["Flight-control interface uses non-blocking UDP telemetry buffers."],
            "matched_kg_requirements": [r["requirement_id"] for r in kg_records],
            "cross_references": [f"{r['standard_source']} {r['requirement_id']}" for r in kg_records],
            "compliance_status": "EVALUATED"
        }

        return final_extracted_data

How the Node Connects to the Graph

In the interpreter we built, the node definition binds this entire sequence cleanly:

# The Node only cares about input state keys, output state keys, and the callable
nodes["ExtractAndQuery"] = FlowNode(
    name="ExtractAndQuery",
    action_key="auditor_action",
    input_keys=["document_chunk", "critique_notes"],
    output_key="extracted_data",
    next_node="EvaluateCritique"
)

# Registered in the engine:
engine.register_action("auditor_action", compliance_auditor_agent.run)

Isolation: The graph engine passes document_chunk and critique_notes from the blackboard into ComplianceAuditorAgent.run().
Scratchpad Lifecycle: The agent creates its local message list, invokes the Neo4j driver, gets the Cypher records, and runs model synthesis.
Commit: The agent returns final_extracted_data. The engine takes this dictionary and writes it to state.extracted_data using the configured reducer. The ephemeral tool-call history and raw database cursors evaporate, leaving only the clean extracted artifact for the downstream ReviewCouncilCritic node.

So this Python tool is additional to the DSL, right?

Yes, exactly.

The DSL is the blueprint; the Python code is the foundation and wiring.

A DSL is purely declarative—it specifies intent, topology, constraints, and schemas, but it cannot execute itself. It needs a host runtime to bring those declarations to life.

The Boundary Between DSL and Host Runtime

┌────────────────────────────────────────────────────────┐
│  DSL LAYER (Declarative Contract)                      │
│                                                        │
│  tool QueryKnowledgeGraph:                             │
│      source: "neo4j://localhost:7687"                  │
│      schema: "schemas/kg_query_input.json"             │
│      timeout_ms: 1500                                  │
│                                                        │
│  agent ComplianceAuditor:                              │
│      model: AuditorEngine                              │
│      tools: [QueryKnowledgeGraph]                      │
└───────────────────────────┬────────────────────────────┘
                            │ Compiles / Binds To
                            ▼
┌────────────────────────────────────────────────────────┐
│  HOST RUNTIME LAYER (Python / Engine Implementation)   │
│                                                        │
│  1. Tool Driver:                                       │
│     - Connects Neo4j Bolt driver                       │
│     - Runs session pools and Cypher query execution    │
│                                                        │
│  2. Agent Harness:                                     │
│     - Translates tool schemas into API function specs  │
│     - Manages LLM tool-calling loop and scratchpad     │
│                                                        │
│  3. State Machine Engine:                              │
│     - Resolves nodes, routes edges, applies reducers   │
└────────────────────────────────────────────────────────┘

Why Keep Them Separate?

Portability of the Workflow: You can take the exact same DSL script and run it against a Python runtime today, or compile it to a high-performance compiled runtime (like Rust using Tokio) tomorrow, without changing the business logic or graph structure.
Security and Sandboxing: You generally don't want arbitrary database drivers, network sockets, or raw SQL/Cypher connection pools defined directly inside workflow text files. The DSL declares which vetted capability is authorized (QueryKnowledgeGraph); the runtime provides the hardened, audited implementation.
Separation of Roles:
- Domain / Systems Engineers author and modify the DSL files (defining agents, critique gates, iteration bounds, and compliance thresholds).
- Software / Infrastructure Engineers write the Python tool adapters, connection pooling, telemetry hooks, and driver bindings.

In formal compiler terminology, the DSL defines the Abstract Syntax Tree (AST), while the Python tool is the foreign function interface (FFI) / native runtime library.

How could a DSL compiler generate the Python boilerplate and Pydantic schemas for tools automatically?

To bridge the declarative DSL and the native Python runtime without manual glue code, the DSL compiler can act as an Ahead-Of-Time (AOT) Code Generator.

The compiler ingests the high-level tool declaration, extracts its operational metadata and parameter contract, and emits:

Strongly typed Pydantic models for validation and OpenAI/Anthropic/Gemini tool-call schema generation.
An Abstract Base Class (ABC) defining the execution interface.
An Engine Harness Adapter that wires the tool directly into the agent’s scratchpad loop.

1. Extended DSL Syntax with Inline Schema Definition

To allow code generation, the DSL must declare parameter types, validation constraints, and descriptions natively:

tool QueryKnowledgeGraph:
    description: "Queries the compliance knowledge graph to retrieve applicable standards and rules."
    source: "neo4j://localhost:7687"
    timeout_ms: 1500
    
    params:
        topic: string(min_length=3, description="Technical topic, e.g., 'telemetry', 'memory_isolation'")
        standard_id: optional[string](regex=r"^[A-Z]+-[A-Z]+-[0-9.]+[A-Z]?$", description="Specific governing standard")
        max_results: int(default=5, ge=1, le=20, description="Maximum number of requirement nodes to return")
        
    returns: list[dict]

2. Code Generation Pipeline

 DSL Tool Block
       │
       ▼
┌──────────────┐
│ Lexer/Parser │ ──► AST: ToolDef(name, params, types, constraints, timeout)
└──────────────┘
       │
       ▼
┌──────────────┐
│  Code-Gen    │ ──► Jinja2 Template / AST Builder
│    Engine    │
└──────────────┘
       │
       ▼
 Generated Python File (`tools/generated_kg_tool.py`)
 ├── Pydantic Input Schema (Auto-generates JSON Schema for LLM)
 ├── Abstract Tool Interface (Forces developer to implement `execute`)
 └── Async Tool Wrapper (Handles timeout, telemetry, schema validation)

3. The Compiler Implementation (Python Script)

Below is a self-contained compiler module that takes the parsed AST structure and renders production-ready Python code:

import os
from dataclasses import dataclass, field
from typing import List, Dict, Optional

# --- Intermediate Representation (IR) from Parser ---

@dataclass
class ParamField:
    name: str
    py_type: str
    is_optional: bool = False
    default: Optional[str] = None
    description: str = ""
    min_length: Optional[int] = None
    regex: Optional[str] = None
    ge: Optional[int] = None
    le: Optional[int] = None

@dataclass
class ToolAST:
    name: str
    description: str
    source_uri: str
    timeout_ms: int
    params: List[ParamField]
    return_type: str

# --- Code Generator ---

class ToolBoilerplateGenerator:
    def __init__(self, tool: ToolAST):
        self.tool = tool

    def render(self) -> str:
        lines = [
            "# AUTO-GENERATED BY AGENTFLOW COMPILER - DO NOT EDIT DIRECTLY",
            "from __future__ import annotations",
            "import abc",
            "import asyncio",
            "from typing import Any, Dict, List, Optional",
            "from pydantic import BaseModel, Field",
            "",
            "# ---------------------------------------------------------------------------",
            "# 1. Strongly Typed Pydantic Schema (Passed to LLM Function Calling)",
            "# ---------------------------------------------------------------------------",
            f"class {self.tool.name}Input(BaseModel):",
            f'    """Input parameters for {self.tool.name}."""'
        ]

        for p in self.tool.params:
            field_args = []
            if p.default is not None:
                field_args.append(f"default={p.default}")
            elif p.is_optional:
                field_args.append("default=None")
            else:
                field_args.append("...")

            if p.description:
                field_args.append(f'description="{p.description}"')
            if p.min_length is not None:
                field_args.append(f"min_length={p.min_length}")
            if p.regex is not None:
                field_args.append(f'pattern="{p.regex}"')
            if p.ge is not None:
                field_args.append(f"ge={p.ge}")
            if p.le is not None:
                field_args.append(f"le={p.le}")

            type_hint = f"Optional[{p.py_type}]" if p.is_optional else p.py_type
            args_str = ", ".join(field_args)
            lines.append(f"    {p.name}: {type_hint} = Field({args_str})")

        lines.extend([
            "",
            "    @classmethod",
            "    def to_openai_tool_schema(cls) -> Dict[str, Any]:",
            "        return {",
            f'            "name": "{self.tool.name}",',
            f'            "description": "{self.tool.description}",',
            '            "parameters": cls.model_json_schema()',
            "        }",
            "",
            "# ---------------------------------------------------------------------------",
            "# 2. Abstract Base Contract (Infrastructure Hook)",
            "# ---------------------------------------------------------------------------",
            f"class Base{self.tool.name}Adapter(abc.ABC):",
            f'    """Abstract executor for {self.tool.name}. Developers implement this."""',
            f"    SOURCE_URI: str = '{self.tool.source_uri}'",
            f"    TIMEOUT_MS: int = {self.tool.timeout_ms}",
            "",
            "    @abc.abstractmethod",
            f"    async def execute(self, params: {self.tool.name}Input) -> {self.tool.return_type}:",
            "        pass",
            "",
            "# ---------------------------------------------------------------------------",
            "# 3. Engine Invocation Wrapper (Lifecycle, Validation, & Timeout Guard)",
            "# ---------------------------------------------------------------------------",
            f"class {self.tool.name}Invoker:",
            f"    def __init__(self, adapter: Base{self.tool.name}Adapter):",
            "        self.adapter = adapter",
            "",
            f"    async def invoke_raw(self, raw_kwargs: Dict[str, Any]) -> {self.tool.return_type}:",
            "        # 1. Validate payload using Pydantic",
            f"        validated_args = {self.tool.name}Input(**raw_kwargs)",
            "",
            "        # 2. Execute with bounded timeout from DSL spec",
            "        try:",
            "            return await asyncio.wait_for(",
            "                self.adapter.execute(validated_args),",
            f"                timeout={self.tool.timeout_ms} / 1000.0",
            "            )",
            "        except asyncio.TimeoutError:",
            f'            raise TimeoutError("Tool {self.tool.name} timed out after {self.tool.timeout_ms}ms")',
            ""
        ])

        return "\n".join(lines)

4. What the Generated Code Looks Like

Running this generator for QueryKnowledgeGraph emits tools/generated_kg_tool.py:

# AUTO-GENERATED BY AGENTFLOW COMPILER - DO NOT EDIT DIRECTLY
from __future__ import annotations
import abc
import asyncio
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field

# ---------------------------------------------------------------------------
# 1. Strongly Typed Pydantic Schema (Passed to LLM Function Calling)
# ---------------------------------------------------------------------------
class QueryKnowledgeGraphInput(BaseModel):
    """Input parameters for QueryKnowledgeGraph."""
    topic: str = Field(..., description="Technical topic, e.g., 'telemetry', 'memory_isolation'", min_length=3)
    standard_id: Optional[str] = Field(default=None, description="Specific governing standard", pattern="^[A-Z]+-[A-Z]+-[0-9.]+[A-Z]?$")
    max_results: int = Field(default=5, description="Maximum number of requirement nodes to return", ge=1, le=20)

    @classmethod
    def to_openai_tool_schema(cls) -> Dict[str, Any]:
        return {
            "name": "QueryKnowledgeGraph",
            "description": "Queries the compliance knowledge graph to retrieve applicable standards and rules.",
            "parameters": cls.model_json_schema()
        }

# ---------------------------------------------------------------------------
# 2. Abstract Base Contract (Infrastructure Hook)
# ---------------------------------------------------------------------------
class BaseQueryKnowledgeGraphAdapter(abc.ABC):
    """Abstract executor for QueryKnowledgeGraph. Developers implement this."""
    SOURCE_URI: str = 'neo4j://localhost:7687'
    TIMEOUT_MS: int = 1500

    @abc.abstractmethod
    async def execute(self, params: QueryKnowledgeGraphInput) -> List[dict]:
        pass

# ---------------------------------------------------------------------------
# 3. Engine Invocation Wrapper (Lifecycle, Validation, & Timeout Guard)
# ---------------------------------------------------------------------------
class QueryKnowledgeGraphInvoker:
    def __init__(self, adapter: BaseQueryKnowledgeGraphAdapter):
        self.adapter = adapter

    async def invoke_raw(self, raw_kwargs: Dict[str, Any]) -> List[dict]:
        validated_args = QueryKnowledgeGraphInput(**raw_kwargs)
        try:
            return await asyncio.wait_for(
                self.adapter.execute(validated_args),
                timeout=1500 / 1000.0
            )
        except asyncio.TimeoutError:
            raise TimeoutError("Tool QueryKnowledgeGraph timed out after 1500ms")

5. All the Developer Has to Write

The human engineer no longer writes boilerplate schema validators, regex checks, or timeout handling. They only implement the concrete adapter:

from neo4j import AsyncGraphDatabase
from generated_kg_tool import (
    BaseQueryKnowledgeGraphAdapter, 
    QueryKnowledgeGraphInput, 
    QueryKnowledgeGraphInvoker
)

class Neo4jProductionAdapter(BaseQueryKnowledgeGraphAdapter):
    def __init__(self, auth: tuple[str, str]):
        self._driver = AsyncGraphDatabase.driver(self.SOURCE_URI, auth=auth)

    async def execute(self, params: QueryKnowledgeGraphInput) -> list[dict]:
        # params is fully validated by Pydantic (types, regex, min_length enforced)
        cypher = """
        MATCH (req:Requirement)-[:GOVERNS_TOPIC]->(t:Topic {name: $topic})
        WHERE (\(standard_id IS NULL OR req.standard_id =\)standard_id)
        RETURN req.id AS requirement_id, req.title AS title
        LIMIT $max_results
        """
        async with self._driver.session() as session:
            result = await session.run(
                cypher, 
                topic=params.topic.lower(), 
                standard_id=params.standard_id, 
                max_results=params.max_results
            )
            return await result.data()

# Wire into the engine
adapter = Neo4jProductionAdapter(auth=("neo4j", "password"))
tool_invoker = QueryKnowledgeGraphInvoker(adapter)

# Register with interpreter:
# engine.register_action("QueryKnowledgeGraph", tool_invoker.invoke_raw)

Operational Benefits

No Type Mismatches: If the DSL author changes max_results upper bound from 20 to 50 in the .aflow file, re-running the compiler updates the Pydantic validator instantly.
Direct LLM Tool Spec Export: QueryKnowledgeGraphInput.to_openai_tool_schema() (or Anthropic/Gemini equivalent via model_json_schema()) feeds the model API directly, eliminating schema hallucination between the prompt and the database.
Deterministic Runtime Guardrails: Timeouts, null safety, and string patterns are enforced at the Python barrier before touching Neo4j or external APIs.

From <https://gemini.google.com/app/92fea2e490a225ef> Google Gemini (3.8 Flash)

Search This Blog

Learning from ChatGPT