Change the ID strategy

By default, agentflow generates UUID v4 strings for thread IDs and run IDs. If your storage backend requires integer primary keys, short human-readable codes, or timestamp-sortable IDs, you can swap in a different generator.

Prerequisites

You have a working graph. No extra packages required.

Quick start: switch to timestamp-based integers

from agentflow.core import StateGraph
from agentflow.utils.id_generator import BigIntIDGenerator

graph = StateGraph(
    id_generator=BigIntIDGenerator(),
)
app = graph.compile()

Every thread and run created by this graph will now have a 19-digit integer ID based on the current nanosecond timestamp.

Available generators

Class	`id_type`	Example output	Use when
`DefaultIDGenerator`	`STRING`	`""` (falls back to framework UUID)	Default — framework picks UUID if empty.
`UUIDGenerator`	`STRING`	`"550e8400-e29b-41d4-a716-446655440000"`	Maximum collision resistance; stateless.
`BigIntIDGenerator`	`BIGINT`	`1712576400000000000`	PostgreSQL `bigint` primary keys; sortable by time.
`TimestampIDGenerator`	`INTEGER`	`1712576400123456`	16-digit microsecond integer; sortable.
`IntIDGenerator`	`INTEGER`	`2147483647`	32-bit random integer; small storage footprint.
`HexIDGenerator`	`STRING`	`"1a2b3c4d5e6f7890abcdef1234567890"`	32-char hex; no hyphens.
`ShortIDGenerator`	`STRING`	`"Ab3XyZ9k"`	Human-readable 8-char codes; URL-safe.

All classes are importable from agentflow.utils.id_generator or the top-level agentflow.utils.

Write a custom generator

from agentflow.utils.id_generator import BaseIDGenerator, IDType
import uuid

class PrefixedUUIDGenerator(BaseIDGenerator):
    """Generates IDs like 'sess_550e8400-e29b-...' for easy prefix filtering."""

    def __init__(self, prefix: str = "sess"):
        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=PrefixedUUIDGenerator("run"))

Async generators

If your ID generation requires I/O (for example, fetching a sequence from a database), implement AsyncIDGenerator:

from agentflow.utils.id_generator import AsyncIDGenerator, IDType

class DatabaseSequenceGenerator(AsyncIDGenerator):
    """Fetch the next integer sequence from PG."""

    def __init__(self, pool):
        self.pool = pool

    @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('agentflow_id_seq')")

Access the generated ID inside a node

The current run's generated ID is available via dependency injection:

from injectq import Inject

async def my_node(
    state,
    config: dict,
    generated_id: str = Inject["generated_id"],
) -> ...:
    print(f"This run ID: {generated_id}")
    return state

Common errors

Error	Cause	Fix
IDs collide in `BigIntIDGenerator` at very high throughput	Two calls land in the same nanosecond.	Switch to `UUIDGenerator` or add a random suffix.
`IntIDGenerator` causes `UNIQUE` constraint failures	32-bit space is too small for your dataset.	Use `BigIntIDGenerator` or `UUIDGenerator`.
`ShortIDGenerator` collisions in production	62^8 ≈ 218 trillion possibilities but not cryptographically guaranteed unique.	Only use for low-volume, human-facing IDs; don't use as a DB primary key.

Prerequisites​

Quick start: switch to timestamp-based integers​

Available generators​

Write a custom generator​

Async generators​

Access the generated ID inside a node​

Common errors​