ID Generator
When to use this
Use an id_generator when your database or external systems require a specific identifier format (bigint primary keys, short codes, prefixed strings, etc.). Pass it to StateGraph(id_generator=...) and the framework uses it everywhere IDs are generated.
Import path
from agentflow.utils.id_generator import (
BaseIDGenerator,
AsyncIDGenerator,
DefaultIDGenerator,
UUIDGenerator,
BigIntIDGenerator,
TimestampIDGenerator,
IntIDGenerator,
HexIDGenerator,
ShortIDGenerator,
IDType,
)
# also available from top-level:
from agentflow.utils import BaseIDGenerator, UUIDGenerator, BigIntIDGenerator # etc.
IDType
Enum returned by BaseIDGenerator.id_type.
| Value | Description |
|---|---|
IDType.STRING | The generator produces str values. |
IDType.INTEGER | The generator produces int values (32-bit range). |
IDType.BIGINT | The generator produces int values (64-bit / timestamp range). |
Built-in generators
DefaultIDGenerator
from agentflow.utils.id_generator import DefaultIDGenerator
graph = StateGraph(id_generator=DefaultIDGenerator())
Returns an empty string. The framework substitutes a UUID when it receives an empty string. Use this when you want the framework default (UUID v4) without explicitly specifying it.
id_type | IDType.STRING |
|---|---|
| Output example | "" → framework substitutes UUID |
UUIDGenerator
UUID version 4, 36-character string.
graph = StateGraph(id_generator=UUIDGenerator())
id_type | IDType.STRING |
|---|---|
| Output example | "550e8400-e29b-41d4-a716-446655440000" |
| Uniqueness | Cryptographically random, collision-resistant |
BigIntIDGenerator
Nanosecond Unix timestamp as a big integer. Time-sortable.
graph = StateGraph(id_generator=BigIntIDGenerator())
id_type | IDType.BIGINT |
|---|---|
| Output example | 1712576400000000000 |
| Use when | PostgreSQL bigint column; need time-ordered IDs |
TimestampIDGenerator
Microsecond Unix timestamp as an integer.
graph = StateGraph(id_generator=TimestampIDGenerator())
id_type | IDType.INTEGER |
|---|---|
| Output example | 1712576400123456 |
| Use when | Sortable IDs with smaller footprint than BigIntIDGenerator |
IntIDGenerator
32-bit cryptographically random integer.
graph = StateGraph(id_generator=IntIDGenerator())
id_type | IDType.INTEGER |
|---|---|
| Output example | 2147483647 |
| Use when | Compact integer IDs; low-volume applications |
HexIDGenerator
32-character hexadecimal string (16 random bytes, no hyphens).
graph = StateGraph(id_generator=HexIDGenerator())
id_type | IDType.STRING |
|---|---|
| Output example | "1a2b3c4d5e6f7890abcdef1234567890" |
| Use when | UUID-like uniqueness without hyphens |
ShortIDGenerator
8-character alphanumeric string (62 possible characters per position).
graph = StateGraph(id_generator=ShortIDGenerator())
id_type | IDType.STRING |
|---|---|
| Output example | "Ab3XyZ9k" |
| Use when | Human-readable IDs in URLs, UI, logs |
| Warning | ~218 trillion combinations — acceptable for low-volume use, not for high-volume DB PKs |
BaseIDGenerator
Abstract base class for synchronous custom generators.
from agentflow.utils.id_generator import BaseIDGenerator, IDType
class PrefixedGenerator(BaseIDGenerator):
def __init__(self, prefix: str = "req"):
self.prefix = prefix
@property
def id_type(self) -> IDType:
return IDType.STRING
def generate(self) -> str:
import uuid
return f"{self.prefix}_{uuid.uuid4()}"
graph = StateGraph(id_generator=PrefixedGenerator("sess"))
Abstract interface
| Member | Description |
|---|---|
id_type (property) | Must implement. Return IDType.STRING, IDType.INTEGER, or IDType.BIGINT. |
generate() | Must implement. Return str | int. May also return Awaitable[str | int] (the framework handles both). |
AsyncIDGenerator
Use when ID generation requires async I/O (e.g., a database sequence).
from agentflow.utils.id_generator import AsyncIDGenerator, IDType
class DBSequenceGenerator(AsyncIDGenerator):
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 generated ID inside a node
The framework binds the generated ID under the key "generated_id". Inject it via Inject["generated_id"]:
from injectq import Inject
async def my_node(
state,
config: dict,
generated_id: str = Inject["generated_id"],
generated_id_type: str = Inject["generated_id_type"], # IDType value
) -> ...:
print(f"Run ID: {generated_id} (type: {generated_id_type})")
return state
Common errors
| Error | Cause | Fix |
|---|---|---|
UNIQUE constraint violation in DB | IntIDGenerator or TimestampIDGenerator collision. | Switch to UUIDGenerator or BigIntIDGenerator. |
| DB column type mismatch | Generator returns int but column is varchar. | Match IDType to your DB schema, or wrap the value with str(). |
generated_id is empty string in storage | Using DefaultIDGenerator and the framework default is not configured. | Explicitly use UUIDGenerator() or another concrete generator. |