ID Generator

An ID generator decides the format of the identifiers 10xGraph creates for threads, runs and messages. Pass one to StateGraph(id_generator=...), or subclass BaseIDGenerator to write your own. This page lists IDType, the base interface, every built-in generator, and how generated IDs reach your nodes.

Import path

from tenxgraph.utils.id_generator import (
    AsyncIDGenerator,
    BaseIDGenerator,
    BigIntIDGenerator,
    DefaultIDGenerator,
    HexIDGenerator,
    IDType,
    IntIDGenerator,
    ShortIDGenerator,
    TimestampIDGenerator,
    UUIDGenerator,
)

Choose a generator

StateGraph takes id_generator: BaseIDGenerator | None = None and falls back to DefaultIDGenerator() when you pass nothing. The prebuilt agents (ReactAgent, SwarmAgent and others) also accept id_generator and forward it to their graph.

IDType

IDType is a str enum that declares what kind of value a generator returns. The framework reads it to pick a fallback when a generator returns an empty value.

The enum is a label, not a validator. The framework does not check that generate() returns a value within a range for the declared type.

BaseIDGenerator

BaseIDGenerator is the abstract base class for all generators. It has two abstract members, and a subclass that omits either cannot be instantiated.

Built-in generators

Every built-in generator takes no constructor arguments. Each section shows the value generate() returns.

DefaultIDGenerator

DefaultIDGenerator returns an empty string, which tells the framework to create the ID itself. This is the generator StateGraph uses when you pass none.

from tenxgraph import StateGraph
from tenxgraph.utils.id_generator import DefaultIDGenerator

# An empty ID makes the framework fall back to a UUID v4 string.
graph = StateGraph(id_generator=DefaultIDGenerator())

UUIDGenerator

UUIDGenerator returns str(uuid.uuid4()), a 36-character string such as "550e8400-e29b-41d4-a716-446655440000". Use it when you want an explicit, always-unique string ID.

from tenxgraph import StateGraph
from tenxgraph.utils.id_generator import UUIDGenerator

graph = StateGraph(id_generator=UUIDGenerator())

AsyncIDGenerator

AsyncIDGenerator is a concrete generator whose generate() is a coroutine that returns a UUID v4 string. It is a reference for the async shape; it performs no real I/O. See Write an async generator for a real one, and read the caveat there first.

HexIDGenerator

HexIDGenerator returns 32 hexadecimal characters built from 16 cryptographically random bytes (secrets.token_hex(16)), for example "1a2b3c4d5e6f7890abcdef1234567890".

from tenxgraph import StateGraph
from tenxgraph.utils.id_generator import HexIDGenerator

graph = StateGraph(id_generator=HexIDGenerator())

ShortIDGenerator

ShortIDGenerator returns 8 characters chosen at random from 62 letters and digits, for example "Ab3XyZ9k". That gives 62^8, about 2.18 x 10^14, combinations.

from tenxgraph import StateGraph
from tenxgraph.utils.id_generator import ShortIDGenerator

graph = StateGraph(id_generator=ShortIDGenerator())

BigIntIDGenerator

BigIntIDGenerator returns int(time.time() * 1_000_000_000), the Unix time in nanoseconds, typically 19 digits. IDs sort by creation time and fit a PostgreSQL bigint column.

from tenxgraph import StateGraph
from tenxgraph.utils.id_generator import BigIntIDGenerator

graph = StateGraph(id_generator=BigIntIDGenerator())

TimestampIDGenerator

TimestampIDGenerator returns int(time.time() * 1000000), the Unix time in microseconds, typically 16 digits. It sorts by creation time like BigIntIDGenerator but is shorter and has the same collision risk.

from tenxgraph import StateGraph
from tenxgraph.utils.id_generator import TimestampIDGenerator

graph = StateGraph(id_generator=TimestampIDGenerator())

The value exceeds 2^32, so it does not fit a 32-bit integer column. Use a bigint column.

IntIDGenerator

IntIDGenerator returns secrets.randbits(32), a random integer from 0 to 4,294,967,295. Collisions become likely at moderate volume, so use it only for small, non-critical workloads.

from tenxgraph import StateGraph
from tenxgraph.utils.id_generator import IntIDGenerator

graph = StateGraph(id_generator=IntIDGenerator())

Write a custom generator

Subclass BaseIDGenerator, implement id_type and generate(), and pass an instance to StateGraph. Use this when you need a prefix, a different alphabet or an external source.

Write a synchronous generator

A synchronous generator defines generate() as a normal method.

# prefixed_ids.py
import uuid

from tenxgraph import StateGraph
from tenxgraph.utils.id_generator import BaseIDGenerator, IDType


class PrefixedIDGenerator(BaseIDGenerator):
    """Generates string IDs such as 'session_<uuid>'."""

    def __init__(self, prefix: str = "run") -> None:
        self.prefix = prefix

    @property
    def id_type(self) -> IDType:
        return IDType.STRING

    def generate(self) -> str:
        return f"{self.prefix}_{uuid.uuid4()}"


graph = StateGraph(id_generator=PrefixedIDGenerator("session"))

Write an async generator

An async generator defines async def generate(), for example to read from a database sequence. The factory that provides IDs to your nodes accepts async methods, and message creation awaits an awaitable ID.

# sequence_ids.py
from tenxgraph import StateGraph
from tenxgraph.utils.id_generator import BaseIDGenerator, IDType


class DatabaseSequenceGenerator(BaseIDGenerator):
    """Takes IDs from a PostgreSQL sequence through an asyncpg pool."""

    def __init__(self, pool) -> None:
        self.pool = pool  # an asyncpg.Pool you created elsewhere

    @property
    def id_type(self) -> IDType:
        return IDType.BIGINT

    async def generate(self) -> int:
        async with self.pool.acquire() as conn:
            return await conn.fetchval("SELECT nextval('my_id_sequence')")


# pool = await asyncpg.create_pool("postgresql://user:pass@host/db")
# graph = StateGraph(id_generator=DatabaseSequenceGenerator(pool))

Read the generated ID in a node

StateGraph binds two keys in its InjectQ container: "generated_id" (a factory that calls your generator each time it is resolved) and "generated_id_type" (the generator’s IDType). A node can request them with Inject.

# nodes.py
from injectq import Inject


async def my_node(
    state,
    config: dict,
    generated_id=Inject["generated_id"],
    generated_id_type=Inject["generated_id_type"],
):
    # IDType is a str enum, so it compares equal to "string", "integer" or "bigint".
    print(f"New ID: {generated_id} (type: {generated_id_type.value})")
    return state

Distributed deployments

For several nodes or workers, the API package provides SnowFlakeIdGenerator in tenxgraph_api.src.app.utils.snowflake_id_generator. It subclasses BaseIDGenerator, returns IDType.BIGINT, has an async generate(), and raises ImportError if the optional snowflakekit package is missing.

Its constructor takes seven optional integers: snowflake_epoch, total_bits, snowflake_time_bits, snowflake_node_bits, snowflake_node_id, snowflake_worker_id and snowflake_worker_bits. Pass none, and it reads SNOWFLAKE_EPOCH, SNOWFLAKE_TOTAL_BITS, SNOWFLAKE_TIME_BITS, SNOWFLAKE_NODE_BITS, SNOWFLAKE_NODE_ID, SNOWFLAKE_WORKER_ID and SNOWFLAKE_WORKER_BITS from the environment. Pass all seven, or none: a partial set is silently ignored. See Configure an ID generator for setup.

Common errors