Skip to content
Junjo AI

Core Concepts

This page breaks down the fundamental building blocks of the Junjo library. Understanding these concepts is key to effectively designing, building, and debugging your workflows.

State

Section titled “State”

What is it? A BaseState is a Pydantic model that defines the data structure for your workflow’s state. It acts as a centralized, type-safe container for all the data that your workflow will operate on.

Key Characteristics:

  • Pydantic-Based: Leverages Pydantic for data validation and type hinting.
  • Immutable in Practice: While the state object itself can be replaced, it is treated as immutable within the workflow. Nodes do not modify the state directly; they request changes through the store.
from junjo import BaseState


class MyWorkflowState(BaseState):
    user_input: str
    processed_data: dict | None = None
    is_complete: bool = False

Store

Section titled “Store”

What is it? A BaseStore is a class that manages the state of a workflow. It holds the BaseState and provides methods (often called “actions”) to update the state in a controlled and predictable manner.

Key Characteristics:

  • State Management: The single source of truth for the workflow’s state.
  • Redux-Inspired: Follows a pattern where state is updated by dispatching actions, ensuring that state changes are explicit and traceable.
  • Concurrency Safe: Uses an asyncio.Lock to ensure that state updates are atomic, preventing race conditions.
from junjo import BaseStore


class MyWorkflowStore(BaseStore[MyWorkflowState]):
    async def set_processed_data(self, data: dict) -> None:
        await self.set_state({"processed_data": data})


    async def mark_as_complete(self) -> None:
        await self.set_state({"is_complete": True})

Node

Section titled “Node”

What is it? A Node represents a single unit of work in your workflow. It’s where your business logic, API calls, or any other operations are executed.

Key Characteristics:

  • Atomic Unit of Work: Each node should have a single, well-defined responsibility.
  • Interacts with the Store: Nodes receive the workflow’s store as an argument to their service method, allowing them to read the current state and dispatch actions to update it.
  • Asynchronous: The service method is an async function, allowing for non-blocking I/O operations.
from junjo import Node


class ProcessDataNode(Node[MyWorkflowStore]):
    async def service(self, store: MyWorkflowStore) -> None:
        state = await store.get_state()
        # Perform some processing on state.user_input
        processed_data = {"result": "some_value"}
        await store.set_processed_data(processed_data)

Edge

Section titled “Edge”

What is it? An Edge defines a directed connection between two nodes in a workflow graph. It represents a potential path of execution.

Key Characteristics:

  • Defines Flow: Edges connect a tail node to a head node, establishing the sequence of operations.
  • Can be Conditional: An edge can have an associated Condition that determines whether the transition from the tail to the head should occur.
from junjo import Edge


edge = Edge(tail=node1, head=node2)

Condition

Section titled “Condition”

What is it? A Condition is a class that contains logic to determine whether an Edge should be traversed.

Key Characteristics:

  • Pure Function of State: A condition’s evaluate method should only depend on the current state of the workflow. It should not have any side effects.
  • Enables Branching: Conditions are the primary mechanism for creating branching logic in your workflows.
from junjo import Condition


class DataIsProcessed(Condition[MyWorkflowState]):
    def evaluate(self, state: MyWorkflowState) -> bool:
        return state.processed_data is not None


edge = Edge(tail=node1, head=node2, condition=DataIsProcessed())

Graph

Section titled “Graph”

What is it? A Graph is a collection of nodes and edges that defines the complete structure of your workflow.

Key Characteristics:

  • Source and Sinks: A graph has a single entry point (source) and one or more explicit terminal nodes (sinks).
  • Defines the Workflow Structure: The graph is a complete representation of all possible paths of execution in your workflow.
  • Compiles To A Structural Snapshot: Graph.compile() returns a canonical compiled view that Junjo uses for validation, serialization, and rendering-oriented graph operations.
  • Immutable Definition Object: After construction, Junjo makes the graph shape immutable so compile-time caching, validation, traversal, telemetry, and rendering all stay aligned with the same graph definition. Graph source, sinks, edges, edge endpoints, edge conditions, and RunConcurrent child membership cannot be reassigned through Junjo’s public graph APIs. Create a new Graph when the workflow shape needs to change.
from junjo import Graph


workflow_graph = Graph(
    source=start_node,
    sinks=[end_node],
    edges=[
        Edge(tail=start_node, head=process_node),
        Edge(tail=process_node, head=end_node, condition=DataIsProcessed())
    ]
)


# Inspect the normalized structural snapshot for debugging or tooling.
compiled = workflow_graph.compile()
print(compiled.source_node_runtime_id)
print(compiled.sink_node_runtime_ids)

Compiled Graph Snapshot

Section titled “Compiled Graph Snapshot”

What is it? A compiled graph snapshot is the canonical structural view that Junjo derives from a single Graph instance. It is exposed through Graph.compile() as a CompiledGraph and is the one graph representation Junjo uses for validation, ordered traversal lookups, serialization, and rendering-oriented tooling.

Why it exists

  • One Structural Source of Truth: Junjo uses one compiled structural view for validation, serialization, and traversal.
  • Stable Graph Identity: The compiled snapshot assigns deterministic graph_structural_id, node_structural_id, and edge_structural_id values so identical graph shapes can be correlated across runs.
  • Explicit Runtime vs Structural Identity: Compiled nodes and edges retain runtime ids for exact execution correlation while also carrying structural ids for graph-shape analysis and telemetry aggregation.

What you get

  • CompiledGraph: the full compiled snapshot for one graph instance
  • CompiledNode: normalized node metadata, including both runtime and structural identity
  • CompiledEdge: normalized edge metadata, including declared edge order and structural identity
compiled = workflow_graph.compile()


print(compiled.graph_structural_id)


for compiled_node in compiled.compiled_nodes:
    print(
        compiled_node.node_runtime_id,
        compiled_node.node_structural_id,
        compiled_node.node_type_name,
    )


for compiled_edge in compiled.compiled_edges:
    print(
        compiled_edge.edge_structural_id,
        compiled_edge.tail_node_runtime_id,
        compiled_edge.head_node_runtime_id,
    )

Runtime Identity vs Structural Identity

Section titled “Runtime Identity vs Structural Identity”

Junjo exposes two different identity layers for graph executables:

  • Runtime identity: identifies the exact executable instance used in one execution. This is what lets telemetry and hooks talk about the specific node, subflow, or concurrent executable that actually ran.
  • Structural identity: identifies the same conceptual graph position across repeated runs of the same graph shape.

This distinction matters because a fresh workflow execution creates a fresh graph. Runtime identities should rotate from run to run, while structural identities should remain stable when the graph shape does not change.

For telemetry and hooks, the most important fields are:

  • executable_runtime_id: the runtime identity of the executable that fired the span or hook
  • executable_structural_id: the stable structural identity of that same executable
  • enclosing_graph_structural_id: the stable structural identity of the compiled graph that contains the executable

This lets Junjo support both:

  • single-run debugging using runtime ids
  • cross-run aggregation using structural ids

Workflow

Section titled “Workflow”

What is it? A Workflow is the main executable component that takes a graph_factory and a store_factory and runs the defined process.

Key Characteristics:

  • Executable: The Workflow class has an execute method that starts the workflow.
  • Manages Execution: It traverses the graph, executing nodes and evaluating conditions, until one of the declared sinks is reached.
  • Isolated Execution: Each call to execute uses the provided factories to create a fresh Graph and Store, ensuring that each execution is isolated and concurrency-safe.
  • Returns a Snapshot: execute returns an ExecutionResult containing the final state and workflow-local execution counts for that run.
  • Retains Failed Run Identity: An admitted failure raises WorkflowExecutionError with the run ID and detached terminal state. The actual Node, Agent, or application failure remains error.__cause__.
  • Reports Snapshot Completeness: error.state_is_terminal distinguishes a complete terminal Store snapshot from last-known recovery state. A separate terminal evidence failure remains available as error.terminalization_error.
  • Preserves Cancellation: An admitted cancellation raises WorkflowCancelledError. It contains the same run identity while remaining an asyncio.CancelledError.
from junjo import Workflow


def create_graph() -> Graph:
    # ... (graph creation logic)
    return workflow_graph


def create_workflow() -> Workflow[MyWorkflowState, MyWorkflowStore]:
    """Factory function to create a new instance of the workflow."""
    return Workflow[MyWorkflowState, MyWorkflowStore](
        name="My First Workflow",
        graph_factory=create_graph,
        store_factory=lambda: MyWorkflowStore(
            initial_state=MyWorkflowState(user_input="hello")
        )
    )


# Create and execute the workflow
sample_workflow = create_workflow()
result = await sample_workflow.execute()
print(result.state.model_dump())

Application code can persist the exact failed run without installing a Hook or querying telemetry:

from junjo import WorkflowExecutionError


try:
    result = await sample_workflow.execute()
except WorkflowExecutionError as error:
    persist_failed_run(error.run_id)
    raise RuntimeError("Application workflow failed") from error

By default, Workflow.execute() validates the freshly created graph before execution begins. This catches invalid terminal topology, unreachable sinks, and reachable non-sink dead ends before any nodes run. For targeted testing or advanced debugging, validation can be disabled per run:

result = await sample_workflow.execute(validate_graph=False)

Passing Parameters to Factories

To provide parameters to your graph_factory or store_factory when you create a Workflow, you can wrap your factory function call in a lambda. This creates a new, argument-less factory that calls your function with the desired parameters when executed.

This is useful for injecting dependencies like configuration objects or API clients into your graph at instantiation time, while preserving concurrency safety.

# Your factory function that requires a dependency
def create_graph_with_dependency(api_key: str) -> Graph:
    # ... setup graph using the api_key
    return Graph(...)


def create_workflow() -> Workflow[MyState, MyStore]:
    # Instantiate the workflow, using a lambda to create the factory
    return Workflow[MyState, MyStore](
        name="configured_workflow",
        graph_factory=lambda: create_graph_with_dependency(
            api_key="your-secret-key"
        ),
        store_factory=lambda: MyStore(initial_state=MyState())
    )


# The workflow can now be executed normally
workflow = create_workflow()
result = await workflow.execute()
print(result.state.model_dump())