10xGraphv0.10.0Search

    Use Callbacks and Hooks

    In shortHook into graph execution to log, validate, trace, and respond to events at two levels

    • 6 min read
    • 11 sections
    • Updated
    • v0.10.0
    • Markdown

    Callbacks and hooks let you observe and control graph execution at critical points. 10xGraph provides two levels: invocation-level callbacks that fire on every LLM, tool, or MCP call, and graph-level lifecycle hooks that fire at structural events like start, end, error, interrupt, and checkpoint. Use them for logging, metrics, tracing, validation, and compliance.

    Goal

    By the end of this guide, you will understand when to use callbacks vs lifecycle hooks, have two working examples, and know how to integrate them into your graph.

    Prerequisites

    • Python 3.12+
    • A compiled 10xGraph agent or graph from /docs/guides/build-a-graph
    • pip install "10xgraph[google-genai]" (or your provider)

    Callbacks vs lifecycle hooks: when to use each

    Invocation-level callbacks fire on every LLM, tool, or MCP call within a node:

    • before_invoke: Validate or modify inputs before they are sent.
    • after_invoke: Log, filter, or modify outputs after they return.
    • on_error: Handle errors that occur during a call and optionally return a recovery value.

    Use these for input/output validation, prompt injection protection, logging at invocation granularity, or modifying tool behavior on-the-fly.

    Graph-level lifecycle hooks fire at structural events of the entire run:

    • on_graph_start: Initialize state, spans, or observability before the first node runs.
    • on_graph_end: Record metrics, send notifications, or finalize after successful completion.
    • on_graph_error: Alert systems or mask sensitive data when the graph fails.
    • on_interrupt / on_resume: Update external task queues or validate approval data.
    • on_checkpoint: Redact sensitive data or replicate state before persistence.
    • on_state_update: Observe every node transition (most granular).

    Use lifecycle hooks for observability, compliance, state persistence policy, and human-in-the-loop coordination.

    Step 1: Set up a callback manager

    Create a CallbackManager, register callbacks and hooks, and pass it to compile():

    Python
    from tenxgraph.core.graph import StateGraph
from tenxgraph.utils import CallbackManager
from tenxgraph.core.state import AgentState, Message

# Create the manager
cbm = CallbackManager()

# Build your graph
builder = StateGraph(AgentState)
# ... add nodes, edges, etc. ...

# Compile with the manager
app = builder.compile(callback_manager=cbm)

    Step 2: Add invocation-level callbacks

    Register callbacks to validate inputs, log outputs, or handle errors on every LLM/tool call.

    Before-invoke: validate or modify input

    Filter or reject inputs before they reach the LLM or tool:

    Python
    from tenxgraph.utils.callbacks import BeforeInvokeCallback, CallbackContext
from tenxgraph.utils import InvocationType

class PromptGuardCallback(BeforeInvokeCallback):
    """Block requests that contain suspicious patterns."""
    
    async def __call__(self, context: CallbackContext, input_data):
        # For AI invocations, input_data is {"state": AgentState, "config": dict}
        for msg in input_data["state"].context or []:
            if "ignore previous" in msg.text().lower():
                raise ValueError("Prompt injection detected")
        return input_data

# Register it
cbm.register_before_invoke(InvocationType.AI, PromptGuardCallback())

    After-invoke: log or modify output

    Inspect or transform outputs after the LLM/tool returns:

    Python
    from tenxgraph.utils.callbacks import AfterInvokeCallback
import logging

logger = logging.getLogger(__name__)

class AuditLogCallback(AfterInvokeCallback):
    """Log every LLM call for compliance."""
    
    async def __call__(self, context: CallbackContext, input_data, output_data):
        logger.info(
            f"[AUDIT] node={context.node_name} type={context.invocation_type} "
            f"function={context.function_name}"
        )
        return output_data

cbm.register_after_invoke(InvocationType.AI, AuditLogCallback())

    On-error: handle or recover from failures

    Catch errors and optionally return a fallback response:

    Python
    from tenxgraph.utils.callbacks import OnErrorCallback
from tenxgraph.core.state import Message

class FallbackOnLLMError(OnErrorCallback):
    """Return a fallback message if the LLM fails."""
    
    async def __call__(self, context: CallbackContext, input_data, error: Exception):
        if "rate_limit" in str(error).lower():
            # Return a recovery message instead of failing
            return Message.text_message(
                "I'm busy. Please try again in a moment.",
                role="assistant"
            )
        # Return None to re-raise the error
        return None

cbm.register_on_error(InvocationType.AI, FallbackOnLLMError())

    Step 3: Add graph-level lifecycle hooks

    Register hooks to observe or control the entire graph run:

    Python
    from tenxgraph.utils.callbacks import GraphLifecycleHook, GraphLifecycleContext
import uuid
import time

class ObservabilityHook(GraphLifecycleHook):
    """Track execution from start to end with metrics."""
    
    async def on_graph_start(self, context: GraphLifecycleContext, state):
        """Initialize trace when the graph starts."""
        trace_id = str(uuid.uuid4())
        state.execution_meta.internal_data["trace_id"] = trace_id
        state.execution_meta.internal_data["start_time"] = time.time()
        print(f"[START] thread={context.thread_id} trace={trace_id}")
        return state
    
    async def on_graph_end(self, context: GraphLifecycleContext, final_state, messages, total_steps):
        """Record final metrics when the graph completes."""
        elapsed = time.time() - final_state.execution_meta.internal_data.get("start_time", 0)
        trace_id = final_state.execution_meta.internal_data.get("trace_id")
        print(
            f"[END] trace={trace_id} steps={total_steps} "
            f"messages={len(messages)} duration={elapsed:.2f}s"
        )
        return None
    
    async def on_graph_error(self, context: GraphLifecycleContext, error: Exception,
                             partial_state, messages, step, node_name):
        """Alert when the graph fails."""
        trace_id = partial_state.execution_meta.internal_data.get("trace_id")
        print(f"[ERROR] trace={trace_id} node={node_name} error={type(error).__name__}: {error}")
        # Mask sensitive fields before persistence
        partial_state.execution_meta.internal_data["error_masked"] = True
        return partial_state, f"Graph failed: {type(error).__name__}"

# Register the hook
cbm.register_lifecycle_hook(ObservabilityHook())

    Step 4: Run your graph and observe

    Invoke the graph normally. Callbacks and hooks fire automatically:

    Python
    # Build and compile your graph with callbacks registered
cbm = CallbackManager()
cbm.register_before_invoke(InvocationType.AI, PromptGuardCallback())
cbm.register_after_invoke(InvocationType.AI, AuditLogCallback())
cbm.register_on_error(InvocationType.AI, FallbackOnLLMError())
cbm.register_lifecycle_hook(ObservabilityHook())

builder = StateGraph(AgentState)
# ... add nodes and edges ...
app = builder.compile(callback_manager=cbm)

# Invoke the graph
result = await app.ainvoke(
    {"messages": [Message.text_message("What is the capital of France?")]},
    config={"thread_id": "user_123", "run_id": "run_456"}
)

print(f"Final messages: {len(result['messages'])}")

    Expected output:

    plaintext
    [START] thread=user_123 trace=a1b2c3d4-...
[AUDIT] node=agent type=ai function=Agent
[END] trace=a1b2c3d4-... steps=1 messages=2 duration=0.45s
Final messages: 2

    Complete example: compliance audit trail

    This example builds a graph that logs every operation and redacts sensitive data before checkpoint:

    Python
    import logging
import re
from tenxgraph.core.state import Message
from tenxgraph.utils import tool, CallbackManager, InvocationType
from tenxgraph.utils.callbacks import (
    GraphLifecycleHook, GraphLifecycleContext,
    BeforeInvokeCallback, AfterInvokeCallback, CallbackContext,
)
from tenxgraph.prebuilt.agent import ReactAgent

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

EMAIL_RE = re.compile(r"\b[\w.-]+@[\w.-]+\.\w+\b")

# Define a simple tool
@tool
def lookup_email(user_id: str) -> str:
    """Look up a user's email by ID."""
    return f"user_{user_id}@example.com"

# Compliance hook: redact PII before persistence
class ComplianceHook(GraphLifecycleHook):
    async def on_checkpoint(self, context: GraphLifecycleContext, state, messages, is_context_trimmed):
        """Redact email addresses before persisting."""
        for msg in messages:
            for block in msg.content:
                if getattr(block, "type", None) == "text":
                    block.text = EMAIL_RE.sub("[REDACTED_EMAIL]", block.text)
        
        logger.info(f"[COMPLIANCE] Checkpointed {len(messages)} messages (redacted)")
        return state, messages

# Audit callback: log every tool call
class ToolAuditCallback(AfterInvokeCallback):
    async def __call__(self, context: CallbackContext, input_data, output_data):
        if context.function_name:
            logger.info(f"[AUDIT] Tool called: {context.function_name}")
        return output_data

# ReactAgent builds the graph (agent node plus tool node) for you
agent = ReactAgent(
    model="gemini/gemini-2.5-flash",
    tools=[lookup_email],
)

# Register callbacks
cbm = CallbackManager()
cbm.register_lifecycle_hook(ComplianceHook())
cbm.register_after_invoke(InvocationType.TOOL, ToolAuditCallback())

app = agent.compile(callback_manager=cbm)

# Run
result = await app.ainvoke(
    {"messages": [Message.text_message("What is the email for user 42?")]},
    config={"thread_id": "audit_demo", "run_id": "run_001"}
)

    Common errors and fixes

    Error Cause Fix
    callbacks never fire callback_manager not passed to compile(). Use graph.compile(callback_manager=cbm) explicitly.
    KeyError or TypeError in a before_invoke callback input_data is a dict, not a message list. AI calls get {"state", "config"}; tool calls get the tool arguments. Read input_data["state"].context for messages.
    on_error callback changes return type An on_error callback returned something other than a Message or None. Non-Message values are ignored with a warning. Return a Message to recover, or None to re-raise.
    lifecycle hook never fires Hook registered after compile(). Register hooks before calling compile(callback_manager=cbm).
    on_graph_error doesn't suppress error Hooks cannot suppress graph errors. Use on_error callbacks for invocation-level recovery.
    callback blocks the graph A callback does slow synchronous work. Class-based callbacks must define async def __call__; avoid blocking I/O inside them.

    Variations and options

    Multiple callbacks of the same type

    Register several callbacks; they execute in order:

    Python
    cbm.register_before_invoke(InvocationType.AI, GuardCallback())
cbm.register_before_invoke(InvocationType.AI, LoggingCallback())
# Both fire before every LLM call, in order

    Selective callback registration

    Callbacks can check context.node_name to apply only to specific nodes:

    Python
    class SelectiveAudit(AfterInvokeCallback):
    async def __call__(self, context: CallbackContext, input_data, output_data):
        if context.node_name == "sensitive_node":
            logger.critical(f"Sensitive operation detected")
        return output_data

    Validators as a simpler alternative

    For message validation, use BaseValidator instead of before-invoke callbacks:

    Python
    from tenxgraph.utils import BaseValidator
from tenxgraph.utils.validators import ValidationError

class SimpleValidator(BaseValidator):
    async def validate(self, messages):
        for msg in messages:
            if "bad_word" in msg.text().lower():
                raise ValidationError("Content policy violation", "policy")
        return True

cbm.register_input_validator(SimpleValidator())

    Combining callbacks and lifecycle hooks

    Use callbacks for invocation-level logic and lifecycle hooks for graph-level coordination:

    Python
    # Callbacks log each LLM call
cbm.register_after_invoke(InvocationType.AI, AuditLogCallback())

# Hooks correlate all calls in one run
cbm.register_lifecycle_hook(ComplianceHook())

app = graph.compile(callback_manager=cbm)
    Last updated for v0.10.0Edit this page on GitHubReport an issue